f Track event emission in channel and emit upon chan resumption

We track whether the event had previously been emitted in `Channel` and
remove it from `internal_funding_created` entirely. Hence, we now
only emit the event after ChannelMonitorUpdate completion, or upon
channel reestablish. This mitigates a race condition where where we
wouldn't persist the event *and* wouldn't regenerate it on restart,
therefore potentially losing it, if async CMU wouldn't complete before
ChannelManager persistence.

Author: tnull

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -1854,7 +1854,6 @@ fn do_during_funding_monitor_fail(confirm_a_first: bool, restore_b_before_conf:
 	let channel_id = OutPoint { txid: funding_created_msg.funding_txid, index: funding_created_msg.funding_output_index }.to_channel_id();
 	nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created_msg);
 	check_added_monitors!(nodes[1], 1);
-	expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
 
 	chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id()));
@@ -1887,6 +1886,9 @@ fn do_during_funding_monitor_fail(confirm_a_first: bool, restore_b_before_conf:
 	nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
 	nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
 	reconnect_nodes(&nodes[0], &nodes[1], (false, confirm_a_first), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
+
+	// But we want to re-emit ChannelPending
+	expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
 	assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
@@ -2901,7 +2903,6 @@ fn do_test_inbound_reload_without_init_mon(use_0conf: bool, lock_commitment: boo
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created_msg);
 	check_added_monitors!(nodes[1], 1);
-	expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
 
 	// nodes[1] happily sends its funding_signed even though its awaiting the persistence of the
 	// initial ChannelMonitor, but it will decline to send its channel_ready even if the funding

lightning/src/ln/channel.rs

@@ -730,6 +730,9 @@ pub(super) struct Channel<Signer: ChannelSigner> {
 	// blinded paths instead of simple scid+node_id aliases.
 	outbound_scid_alias: u64,
 
+	// We track whether we already emitted a `ChannelPending` event.
+	channel_pending_event_emitted: bool,
+
 	// We track whether we already emitted a `ChannelReady` event.
 	channel_ready_event_emitted: bool,
 
@@ -1107,6 +1110,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			latest_inbound_scid_alias: None,
 			outbound_scid_alias,
 
+			channel_pending_event_emitted: false,
 			channel_ready_event_emitted: false,
 
 			#[cfg(any(test, fuzzing))]
@@ -1456,6 +1460,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 			latest_inbound_scid_alias: None,
 			outbound_scid_alias,
 
+			channel_pending_event_emitted: false,
 			channel_ready_event_emitted: false,
 
 			#[cfg(any(test, fuzzing))]
@@ -4706,6 +4711,16 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 		self.prev_config.map(|prev_config| prev_config.0)
 	}
 
+	// Checks whether we should emit a `ChannelPending` event.
+	pub(crate) fn should_emit_channel_pending_event(&mut self) -> bool {
+		self.is_funding_initiated() && !self.channel_pending_event_emitted
+	}
+
+	// Remembers that we already emitted a `ChannelPending` event.
+	pub(crate) fn set_channel_pending_event_emitted(&mut self) {
+		self.channel_pending_event_emitted = true;
+	}
+
 	// Checks whether we should emit a `ChannelReady` event.
 	pub(crate) fn should_emit_channel_ready_event(&mut self) -> bool {
 		self.is_usable() && !self.channel_ready_event_emitted
@@ -6432,6 +6447,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Writeable for Channel<Signer> {
 			if self.holder_max_htlc_value_in_flight_msat != Self::get_holder_max_htlc_value_in_flight_msat(self.channel_value_satoshis, &old_max_in_flight_percent_config)
 			{ Some(self.holder_max_htlc_value_in_flight_msat) } else { None };
 
+		let channel_pending_event_emitted = Some(self.channel_pending_event_emitted);
 		let channel_ready_event_emitted = Some(self.channel_ready_event_emitted);
 
 		// `user_id` used to be a single u64 value. In order to remain backwards compatible with
@@ -6465,6 +6481,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Writeable for Channel<Signer> {
 			(25, user_id_high_opt, option),
 			(27, self.channel_keys_id, required),
 			(29, self.temporary_channel_id, option),
+			(31, channel_pending_event_emitted, option),
 		});
 
 		Ok(())
@@ -6732,6 +6749,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 		let mut announcement_sigs_state = Some(AnnouncementSigsState::NotSent);
 		let mut latest_inbound_scid_alias = None;
 		let mut outbound_scid_alias = None;
+		let mut channel_pending_event_emitted = None;
 		let mut channel_ready_event_emitted = None;
 
 		let mut user_id_high_opt: Option<u64> = None;
@@ -6758,6 +6776,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 			(25, user_id_high_opt, option),
 			(27, channel_keys_id, option),
 			(29, temporary_channel_id, option),
+			(31, channel_pending_event_emitted, option),
 		});
 
 		let (channel_keys_id, holder_signer) = if let Some(channel_keys_id) = channel_keys_id {
@@ -6915,6 +6934,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 			// Later in the ChannelManager deserialization phase we scan for channels and assign scid aliases if its missing
 			outbound_scid_alias: outbound_scid_alias.unwrap_or(0),
 
+			channel_pending_event_emitted: channel_pending_event_emitted.unwrap_or(true),
 			channel_ready_event_emitted: channel_ready_event_emitted.unwrap_or(true),
 
 			#[cfg(any(test, fuzzing))]

lightning/src/ln/channelmanager.rs

@@ -4249,7 +4249,9 @@ where
 		if let Some(tx) = funding_broadcastable {
 			log_info!(self.logger, "Broadcasting funding transaction with txid {}", tx.txid());
 			self.tx_broadcaster.broadcast_transaction(&tx);
+		}
 
+		if channel.should_emit_channel_pending_event() {
 			let mut pending_events = self.pending_events.lock().unwrap();
 			pending_events.push(events::Event::ChannelPending {
 				channel_id: channel.channel_id(),
@@ -4258,6 +4260,7 @@ where
 				user_channel_id: channel.get_user_id(),
 				funding_txo: channel.get_funding_txo().unwrap().into_bitcoin_outpoint(),
 			});
+			channel.set_channel_pending_event_emitted();
 		}
 
 		emit_channel_ready_event!(self, channel);
@@ -4600,10 +4603,7 @@ where
 					msg: funding_msg,
 				});
 
-				let funding_txo = monitor.get_funding_txo().0;
-				let user_channel_id = chan.get_user_id();
-
-				let monitor_res = self.chain_monitor.watch_channel(funding_txo, monitor);
+				let monitor_res = self.chain_monitor.watch_channel(monitor.get_funding_txo().0, monitor);
 
 				let chan = e.insert(chan);
 				let mut res = handle_new_monitor_update!(self, monitor_res, 0, peer_state_lock, peer_state,
@@ -4618,15 +4618,6 @@ where
 				// with the funding_signed so the channel can never go on chain).
 				if let Err(MsgHandleErrInternal { shutdown_finish: Some((res, _)), .. }) = &mut res {
 					res.0 = None;
-				} else {
-					let mut pending_events = self.pending_events.lock().unwrap();
-					pending_events.push(events::Event::ChannelPending {
-						channel_id: new_channel_id,
-						former_temporary_channel_id: Some(msg.temporary_channel_id),
-						counterparty_node_id: *counterparty_node_id,
-						user_channel_id,
-						funding_txo: funding_txo.into_bitcoin_outpoint(),
-					});
 				}
 				res
 			}

lightning/src/ln/priv_short_conf_tests.rs

@@ -597,12 +597,12 @@ fn test_0conf_channel_with_async_monitor() {
 
 	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
 	nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created);
-	expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
 	check_added_monitors!(nodes[1], 1);
 	assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
 
 	let channel_id = funding_output.to_channel_id();
 	nodes[1].chain_monitor.complete_sole_pending_chan_update(&channel_id);
+	expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id());
 
 	let bs_signed_locked = nodes[1].node.get_and_clear_pending_msg_events();
 	assert_eq!(bs_signed_locked.len(), 2);