Add ChannelClosed generation at cooperative/force-close/error processing

When we detect a channel `is_shutdown()` or call on it
`force_shutdown()`, we notify the user with a Event::ChannelClosed
informing about the id and closure reason.

Author: Antoine Riard

lightning-background-processor/src/lib.rs

@@ -614,6 +614,7 @@ mod tests {
 			.expect("SpendableOutputs not handled within deadline");
 		match event {
 			Event::SpendableOutputs { .. } => {},
+			Event::ChannelClosed { .. } => {},
 			_ => panic!("Unexpected event: {:?}", event),
 		}
 

lightning-persister/src/lib.rs

@@ -182,11 +182,11 @@ mod tests {
 	use bitcoin::Txid;
 	use lightning::chain::channelmonitor::{Persist, ChannelMonitorUpdateErr};
 	use lightning::chain::transaction::OutPoint;
-	use lightning::{check_closed_broadcast, check_added_monitors};
+	use lightning::{check_closed_broadcast, check_closed_event, check_added_monitors};
 	use lightning::ln::features::InitFeatures;
 	use lightning::ln::functional_test_utils::*;
 	use lightning::ln::msgs::ErrorAction;
-	use lightning::util::events::{MessageSendEventsProvider, MessageSendEvent};
+	use lightning::util::events::{ClosureReason, Event, MessageSendEventsProvider, MessageSendEvent};
 	use lightning::util::test_utils;
 	use std::fs;
 	#[cfg(target_os = "windows")]
@@ -259,6 +259,7 @@ mod tests {
 		// Force close because cooperative close doesn't result in any persisted
 		// updates.
 		nodes[0].node.force_close_channel(&nodes[0].node.list_channels()[0].channel_id).unwrap();
+		check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed);
 		check_closed_broadcast!(nodes[0], true);
 		check_added_monitors!(nodes[0], 1);
 
@@ -268,6 +269,7 @@ mod tests {
 		let header = BlockHeader { version: 0x20000000, prev_blockhash: nodes[0].best_block_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 		connect_block(&nodes[1], &Block { header, txdata: vec![node_txn[0].clone(), node_txn[0].clone()]});
 		check_closed_broadcast!(nodes[1], true);
+		check_closed_event!(nodes[1], 1, ClosureReason::CommitmentTxConfirmed);
 		check_added_monitors!(nodes[1], 1);
 
 		// Make sure everything is persisted as expected after close.
@@ -291,6 +293,7 @@ mod tests {
 		let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
 		let chan = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
 		nodes[1].node.force_close_channel(&chan.2).unwrap();
+		check_closed_event!(nodes[1], 1, ClosureReason::HolderForceClosed);
 		let mut added_monitors = nodes[1].chain_monitor.added_monitors.lock().unwrap();
 
 		// Set the persister's directory to read-only, which should result in

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -28,7 +28,7 @@ use ln::msgs::{ChannelMessageHandler, ErrorAction, RoutingMessageHandler};
 use routing::router::get_route;
 use util::config::UserConfig;
 use util::enforcing_trait_impls::EnforcingSigner;
-use util::events::{Event, MessageSendEvent, MessageSendEventsProvider, PaymentPurpose};
+use util::events::{Event, MessageSendEvent, MessageSendEventsProvider, PaymentPurpose, ClosureReason};
 use util::errors::APIError;
 use util::ser::{ReadableArgs, Writeable};
 use util::test_utils::TestBroadcaster;
@@ -81,6 +81,7 @@ fn do_test_simple_monitor_permanent_update_fail(persister_fail: bool) {
 	// PaymentFailed event
 
 	assert_eq!(nodes[0].node.list_channels().len(), 0);
+	check_closed_event!(nodes[0], 1, ClosureReason::ProcessingError { err: "ChannelMonitor storage failure".to_string() });
 }
 
 #[test]
@@ -269,6 +270,7 @@ fn do_test_simple_monitor_temporary_update_fail(disconnect: bool, persister_fail
 	// PaymentFailed event
 
 	assert_eq!(nodes[0].node.list_channels().len(), 0);
+	check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed);
 }
 
 #[test]
@@ -1985,6 +1987,8 @@ fn do_during_funding_monitor_fail(confirm_a_first: bool, restore_b_before_conf:
 
 	send_payment(&nodes[0], &[&nodes[1]], 8000000);
 	close_channel(&nodes[0], &nodes[1], &channel_id, funding_tx, true);
+	check_closed_event!(nodes[0], 1, ClosureReason::CooperativeClosure);
+	check_closed_event!(nodes[1], 1, ClosureReason::CooperativeClosure);
 }
 
 #[test]
@@ -2610,6 +2614,8 @@ fn test_temporary_error_during_shutdown() {
 	assert_eq!(txn_a, txn_b);
 	assert_eq!(txn_a.len(), 1);
 	check_spends!(txn_a[0], funding_tx);
+	check_closed_event!(nodes[1], 1, ClosureReason::CooperativeClosure);
+	check_closed_event!(nodes[0], 1, ClosureReason::CooperativeClosure);
 }
 
 #[test]
@@ -2630,6 +2636,7 @@ fn test_permanent_error_during_sending_shutdown() {
 	assert!(nodes[0].node.close_channel(&channel_id).is_ok());
 	check_closed_broadcast!(nodes[0], true);
 	check_added_monitors!(nodes[0], 2);
+	check_closed_event!(nodes[0], 1, ClosureReason::ProcessingError { err: "ChannelMonitor storage failure".to_string() });
 }
 
 #[test]
@@ -2652,6 +2659,7 @@ fn test_permanent_error_during_handling_shutdown() {
 	nodes[1].node.handle_shutdown(&nodes[0].node.get_our_node_id(), &InitFeatures::known(), &shutdown);
 	check_closed_broadcast!(nodes[1], true);
 	check_added_monitors!(nodes[1], 2);
+	check_closed_event!(nodes[1], 1, ClosureReason::ProcessingError { err: "ChannelMonitor storage failure".to_string() });
 }
 
 #[test]

lightning/src/ln/channelmanager.rs

@@ -52,7 +52,7 @@ use ln::onion_utils;
 use ln::msgs::{ChannelMessageHandler, DecodeError, LightningError, OptionalField};
 use chain::keysinterface::{Sign, KeysInterface, KeysManager, InMemorySigner};
 use util::config::UserConfig;
-use util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider};
+use util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason};
 use util::{byte_utils, events};
 use util::ser::{Readable, ReadableArgs, MaybeReadable, Writeable, Writer};
 use util::chacha20::{ChaCha20, ChaChaReader};
@@ -835,6 +835,9 @@ macro_rules! handle_error {
 							msg: update
 						});
 					}
+					if let Some(channel_id) = chan_id {
+						$self.pending_events.lock().unwrap().push(events::Event::ChannelClosed { channel_id,  reason: ClosureReason::ProcessingError { err: err.err.clone() } });
+					}
 				}
 
 				log_error!($self.logger, "{}", err.err);
@@ -1368,6 +1371,12 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 								msg: channel_update
 							});
 						}
+						if let Ok(mut pending_events_lock) = self.pending_events.lock() {
+							pending_events_lock.push(events::Event::ChannelClosed {
+								channel_id: *channel_id,
+								reason: ClosureReason::HolderForceClosed
+							});
+						}
 					}
 					break Ok(());
 				},
@@ -1443,7 +1452,9 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 		}
 	}
 
-	fn force_close_channel_with_peer(&self, channel_id: &[u8; 32], peer_node_id: Option<&PublicKey>) -> Result<PublicKey, APIError> {
+	/// `peer_node_id` should be set when we receive a message from a peer, but not set when the
+	/// user closes, which will be re-exposed as the `ChannelClosed` reason.
+	fn force_close_channel_with_peer(&self, channel_id: &[u8; 32], peer_node_id: Option<&PublicKey>, peer_msg: Option<&String>) -> Result<PublicKey, APIError> {
 		let mut chan = {
 			let mut channel_state_lock = self.channel_state.lock().unwrap();
 			let channel_state = &mut *channel_state_lock;
@@ -1456,6 +1467,14 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 				if let Some(short_id) = chan.get().get_short_channel_id() {
 					channel_state.short_to_id.remove(&short_id);
 				}
+				let mut pending_events_lock = self.pending_events.lock().unwrap();
+				if peer_node_id.is_some() {
+					if let Some(peer_msg) = peer_msg {
+						pending_events_lock.push(events::Event::ChannelClosed { channel_id: *channel_id, reason: ClosureReason::CounterpartyForceClosed { peer_msg: peer_msg.to_string() } });
+					}
+				} else {
+					pending_events_lock.push(events::Event::ChannelClosed { channel_id: *channel_id, reason: ClosureReason::HolderForceClosed });
+				}
 				chan.remove_entry().1
 			} else {
 				return Err(APIError::ChannelUnavailable{err: "No such channel".to_owned()});
@@ -1477,7 +1496,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 	/// the chain and rejecting new HTLCs on the given channel. Fails if channel_id is unknown to the manager.
 	pub fn force_close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> {
 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
-		match self.force_close_channel_with_peer(channel_id, None) {
+		match self.force_close_channel_with_peer(channel_id, None, None) {
 			Ok(counterparty_node_id) => {
 				self.channel_state.lock().unwrap().pending_msg_events.push(
 					events::MessageSendEvent::HandleError {
@@ -2421,6 +2440,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 											if let Some(short_id) = channel.get_short_channel_id() {
 												channel_state.short_to_id.remove(&short_id);
 											}
+											// ChannelClosed event is generated by handle_error for us.
 											Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel_id, channel.force_shutdown(true), self.get_channel_update_for_broadcast(&channel).ok()))
 										},
 										ChannelError::CloseDelayBroadcast(_) => { panic!("Wait is only generated on receipt of channel_reestablish, which is handled by try_chan_entry, we don't bother to support it here"); }
@@ -3550,6 +3570,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 					msg: update
 				});
 			}
+			self.pending_events.lock().unwrap().push(events::Event::ChannelClosed { channel_id: msg.channel_id,  reason: ClosureReason::CooperativeClosure });
 		}
 		Ok(())
 	}
@@ -3961,6 +3982,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 								msg: update
 							});
 						}
+						self.pending_events.lock().unwrap().push(events::Event::ChannelClosed { channel_id: chan.channel_id(),  reason: ClosureReason::CommitmentTxConfirmed });
 						pending_msg_events.push(events::MessageSendEvent::HandleError {
 							node_id: chan.get_counterparty_node_id(),
 							action: msgs::ErrorAction::SendErrorMessage {
@@ -4022,6 +4044,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 					Err(e) => {
 						let (close_channel, res) = convert_chan_err!(self, e, short_to_id, chan, channel_id);
 						handle_errors.push((chan.get_counterparty_node_id(), Err(res)));
+						// ChannelClosed event is generated by handle_error for us
 						!close_channel
 					}
 				}
@@ -4075,6 +4098,13 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 								});
 							}
 
+							if let Ok(mut pending_events_lock) = self.pending_events.lock() {
+								pending_events_lock.push(events::Event::ChannelClosed {
+									channel_id: *channel_id,
+									reason: ClosureReason::CooperativeClosure
+								});
+							}
+
 							log_info!(self.logger, "Broadcasting {}", log_tx!(tx));
 							self.tx_broadcaster.broadcast_transaction(&tx);
 							false
@@ -4495,6 +4525,7 @@ where
 							msg: update
 						});
 					}
+					self.pending_events.lock().unwrap().push(events::Event::ChannelClosed { channel_id: channel.channel_id(),  reason: ClosureReason::CommitmentTxConfirmed });
 					pending_msg_events.push(events::MessageSendEvent::HandleError {
 						node_id: channel.get_counterparty_node_id(),
 						action: msgs::ErrorAction::SendErrorMessage { msg: e },
@@ -4685,6 +4716,7 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 								msg: update
 							});
 						}
+						self.pending_events.lock().unwrap().push(events::Event::ChannelClosed { channel_id: chan.channel_id(),  reason: ClosureReason::DisconnectedPeer });
 						false
 					} else {
 						true
@@ -4699,6 +4731,7 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 							if let Some(short_id) = chan.get_short_channel_id() {
 								short_to_id.remove(&short_id);
 							}
+							self.pending_events.lock().unwrap().push(events::Event::ChannelClosed { channel_id: chan.channel_id(),  reason: ClosureReason::DisconnectedPeer });
 							return false;
 						} else {
 							no_channels_remain = false;
@@ -4789,12 +4822,12 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 			for chan in self.list_channels() {
 				if chan.counterparty.node_id == *counterparty_node_id {
 					// Untrusted messages from peer, we throw away the error if id points to a non-existent channel
-					let _ = self.force_close_channel_with_peer(&chan.channel_id, Some(counterparty_node_id));
+					let _ = self.force_close_channel_with_peer(&chan.channel_id, Some(counterparty_node_id), Some(&msg.data));
 				}
 			}
 		} else {
 			// Untrusted messages from peer, we throw away the error if id points to a non-existent channel
-			let _ = self.force_close_channel_with_peer(&msg.channel_id, Some(counterparty_node_id));
+			let _ = self.force_close_channel_with_peer(&msg.channel_id, Some(counterparty_node_id), Some(&msg.data));
 		}
 	}
 }
@@ -5295,6 +5328,7 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 		let mut funding_txo_set = HashSet::with_capacity(cmp::min(channel_count as usize, 128));
 		let mut by_id = HashMap::with_capacity(cmp::min(channel_count as usize, 128));
 		let mut short_to_id = HashMap::with_capacity(cmp::min(channel_count as usize, 128));
+		let mut channel_closures = Vec::new();
 		for _ in 0..channel_count {
 			let mut channel: Channel<Signer> = Channel::read(reader, &args.keys_manager)?;
 			let funding_txo = channel.get_funding_txo().ok_or(DecodeError::InvalidValue)?;
@@ -5325,6 +5359,10 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 					let (_, mut new_failed_htlcs) = channel.force_shutdown(true);
 					failed_htlcs.append(&mut new_failed_htlcs);
 					monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger);
+					channel_closures.push(events::Event::ChannelClosed {
+						channel_id: channel.channel_id(),
+						reason: ClosureReason::OutdatedChannelManager
+					});
 				} else {
 					if let Some(short_channel_id) = channel.get_short_channel_id() {
 						short_to_id.insert(short_channel_id, channel.channel_id());
@@ -5432,6 +5470,10 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 		let mut secp_ctx = Secp256k1::new();
 		secp_ctx.seeded_randomize(&args.keys_manager.get_secure_random_bytes());
 
+		if !channel_closures.is_empty() {
+			pending_events_read.append(&mut channel_closures);
+		}
+
 		let channel_manager = ChannelManager {
 			genesis_hash,
 			fee_estimator: args.fee_estimator,

lightning/src/ln/functional_test_utils.rs

@@ -743,16 +743,16 @@ macro_rules! get_closing_signed_broadcast {
 #[macro_export]
 macro_rules! check_closed_broadcast {
 	($node: expr, $with_error_msg: expr) => {{
-		let events = $node.node.get_and_clear_pending_msg_events();
-		assert_eq!(events.len(), if $with_error_msg { 2 } else { 1 });
-		match events[0] {
+		let msg_events = $node.node.get_and_clear_pending_msg_events();
+		assert_eq!(msg_events.len(), if $with_error_msg { 2 } else { 1 });
+		match msg_events[0] {
 			MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
 				assert_eq!(msg.contents.flags & 2, 2);
 			},
 			_ => panic!("Unexpected event"),
 		}
 		if $with_error_msg {
-			match events[1] {
+			match msg_events[1] {
 				MessageSendEvent::HandleError { action: ErrorAction::SendErrorMessage { ref msg }, node_id: _ } => {
 					// TODO: Check node_id
 					Some(msg.clone())
@@ -763,6 +763,24 @@ macro_rules! check_closed_broadcast {
 	}}
 }
 
+/// Check that a channel's closing channel event has been issued
+#[macro_export]
+macro_rules! check_closed_event {
+	($node: expr, $events: expr, $reason: expr) => {{
+		let events = $node.node.get_and_clear_pending_events();
+		assert_eq!(events.len(), $events);
+		let expected_reason = $reason;
+		for event in events {
+			match event {
+				Event::ChannelClosed { ref reason, .. } => {
+					assert_eq!(*reason, expected_reason);
+				},
+				_ => panic!("Unexpected event"),
+			}
+		}
+	}}
+}
+
 pub fn close_channel<'a, 'b, 'c>(outbound_node: &Node<'a, 'b, 'c>, inbound_node: &Node<'a, 'b, 'c>, channel_id: &[u8; 32], funding_tx: Transaction, close_inbound_first: bool) -> (msgs::ChannelUpdate, msgs::ChannelUpdate, Transaction) {
 	let (node_a, broadcaster_a, struct_a) = if close_inbound_first { (&inbound_node.node, &inbound_node.tx_broadcaster, inbound_node) } else { (&outbound_node.node, &outbound_node.tx_broadcaster, outbound_node) };
 	let (node_b, broadcaster_b, struct_b) = if close_inbound_first { (&outbound_node.node, &outbound_node.tx_broadcaster, outbound_node) } else { (&inbound_node.node, &inbound_node.tx_broadcaster, inbound_node) };
@@ -986,6 +1004,20 @@ macro_rules! expect_pending_htlcs_forwardable {
 	}}
 }
 
+#[cfg(test)]
+macro_rules! expect_pending_htlcs_forwardable_from_events {
+	($node: expr, $events: expr, $ignore: expr) => {{
+		assert_eq!($events.len(), 1);
+		match $events[0] {
+			Event::PendingHTLCsForwardable { .. } => { },
+			_ => panic!("Unexpected event"),
+		};
+		if $ignore {
+			$node.node.process_pending_htlc_forwards();
+		}
+	}}
+}
+
 #[cfg(any(test, feature = "unstable"))]
 macro_rules! expect_payment_received {
 	($node: expr, $expected_payment_hash: expr, $expected_payment_secret: expr, $expected_recv_value: expr) => {