Handle events immediately if we are running during block connection

During block connection, we cannot apply `ChannelMonitorUpdate`s if
we're running during the startup sequence (i.e. before the user has
called any methods outside of block connection). We previously
handled this by simply always pushing any `ChannelMonitorUpdate`s
generated during block connection into the
`pending_background_events` queue.

However, this results in `ChannelMonitorUpdate`s going through the
queue when we could just push them immediately. Here we explicitly
check `background_events_processed_since_startup` and use that to
decide whether to push updates through the background queue
instead.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -9624,8 +9624,8 @@ where
 	}
 
 	/// Handle a list of channel failures during a block_connected or block_disconnected call,
-	/// pushing the channel monitor update (if any) to the background events queue and removing the
-	/// Channel object.
+	/// pushing the channel monitor update (if any) to the background events queue (if we're
+	/// currently in the startup phase) and calling [`Self::finish_close_channel`].
 	fn handle_init_event_channel_failures(&self, mut failed_channels: Vec<ShutdownResult>) {
 		for mut failure in failed_channels.drain(..) {
 			// Either a commitment transactions has been confirmed on-chain or
@@ -9640,10 +9640,16 @@ where
 				if let ChannelMonitorUpdateStep::ChannelForceClosed { should_broadcast } = update.updates[0] {
 					assert!(should_broadcast);
 				} else { unreachable!(); }
-				self.pending_background_events.lock().unwrap().push(
-					BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
-						counterparty_node_id, funding_txo, update, channel_id,
-					});
+				if self.background_events_processed_since_startup.load(Ordering::Acquire) {
+					let res = self.chain_monitor.update_channel(funding_txo, &update);
+					debug_assert_eq!(res, ChannelMonitorUpdateStatus::Completed,
+						"TODO: We don't currently handle failures here, this logic is removed in the next commit");
+				} else {
+					self.pending_background_events.lock().unwrap().push(
+						BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
+							counterparty_node_id, funding_txo, update, channel_id,
+						});
+				}
 			}
 			self.finish_close_channel(failure);
 		}

lightning/src/ln/reorg_tests.rs

@@ -268,6 +268,9 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 	assert_eq!(nodes[1].node.list_channels()[0].confirmations, Some(10));
 
 	if !reorg_after_reload {
+		// With expect_channel_force_closed set the TestChainMonitor will enforce that the next update
+		// is a ChannelForceClosed on the right channel with should_broadcast set.
+		*nodes[0].chain_monitor.expect_channel_force_closed.lock().unwrap() = Some((chan.2, true));
 		if use_funding_unconfirmed {
 			let relevant_txids = nodes[0].node.get_relevant_txids();
 			assert_eq!(relevant_txids.len(), 1);
@@ -293,12 +296,17 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 		let relevant_txids = nodes[0].node.get_relevant_txids();
 		assert_eq!(relevant_txids.len(), 0);
 
+		let txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
+		assert_eq!(txn.len(), 1);
+
 		{
 			let per_peer_state = nodes[0].node.per_peer_state.read().unwrap();
 			let peer_state = per_peer_state.get(&nodes[1].node.get_our_node_id()).unwrap().lock().unwrap();
 			assert_eq!(peer_state.channel_by_id.len(), 0);
 			assert_eq!(nodes[0].node.short_to_chan_info.read().unwrap().len(), 0);
 		}
+
+		check_added_monitors!(nodes[0], 1);
 	}
 
 	if reload_node {
@@ -310,10 +318,13 @@ 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_default_configuration(), &nodes_0_serialized, &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized);
-		assert!(nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().is_empty());
 	}
 
 	if reorg_after_reload {
+		// With expect_channel_force_closed set the TestChainMonitor will enforce that the next update
+		// is a ChannelForceClosed on the right channel with should_broadcast set.
+		*nodes[0].chain_monitor.expect_channel_force_closed.lock().unwrap() = Some((chan.2, true));
+
 		if use_funding_unconfirmed {
 			let relevant_txids = nodes[0].node.get_relevant_txids();
 			assert_eq!(relevant_txids.len(), 1);
@@ -345,12 +356,18 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 			assert_eq!(peer_state.channel_by_id.len(), 0);
 			assert_eq!(nodes[0].node.short_to_chan_info.read().unwrap().len(), 0);
 		}
+
+		if reload_node {
+			// The update may come when we free background events if we just restarted, or in-line if
+			// we were already running.
+			nodes[0].node.test_process_background_events();
+		}
+		check_added_monitors!(nodes[0], 1);
+
+		let txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
+		assert_eq!(txn.len(), 1);
 	}
-	// With expect_channel_force_closed set the TestChainMonitor will enforce that the next update
-	// is a ChannelForcClosed on the right channel with should_broadcast set.
-	*nodes[0].chain_monitor.expect_channel_force_closed.lock().unwrap() = Some((chan.2, true));
-	nodes[0].node.test_process_background_events(); // Required to free the pending background monitor update
-	check_added_monitors!(nodes[0], 1);
+
 	let expected_err = "Funding transaction was un-confirmed. Locked at 6 confs, now have 0 confs.";
 	if reorg_after_reload || !reload_node {
 		handle_announce_close_broadcast_events(&nodes, 0, 1, true, "Channel closed because of an exception: Funding transaction was un-confirmed. Locked at 6 confs, now have 0 confs.");
@@ -361,8 +378,6 @@ fn do_test_unconf_chan(reload_node: bool, reorg_after_reload: bool, use_funding_
 
 	check_closed_event!(nodes[0], 1, ClosureReason::ProcessingError { err: expected_err.to_owned() },
 		[nodes[1].node.get_our_node_id()], 100000);
-	assert_eq!(nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().len(), 1);
-	nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().clear();
 
 	// Now check that we can create a new channel
 	if reload_node && nodes[0].node.per_peer_state.read().unwrap().len() == 0 {