Wake the background processor if an async monitor update completes

If the `ChainMonitor` gets an async monitor update completion, this
means the `ChannelManager` needs to be polled for event processing.
Here we wake it using the new multi-`Future`-await `Sleeper`, or
the existing `select` block in the async BP.

Fixes #2052.

Author: TheBlueMatt

lightning-background-processor/src/lib.rs

@@ -38,6 +38,8 @@ use lightning::routing::router::Router;
 use lightning::routing::scoring::{Score, WriteableScore};
 use lightning::util::logger::Logger;
 use lightning::util::persist::Persister;
+#[cfg(feature = "std")]
+use lightning::util::wakers::Sleeper;
 use lightning_rapid_gossip_sync::RapidGossipSync;
 
 use core::ops::Deref;
@@ -471,7 +473,8 @@ where
 		channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
 		gossip_sync, peer_manager, logger, scorer, should_break, {
 			select_biased! {
-				_ = channel_manager.get_persistable_update_future().fuse() => true,
+				_ = chain_monitor.get_update_future().fuse() => true,
+				_ = channel_manager.get_persistable_update_future().fuse() => false,
 				exit = sleeper(Duration::from_millis(100)).fuse() => {
 					should_break = exit;
 					false
@@ -597,7 +600,10 @@ impl BackgroundProcessor {
 			define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
 				channel_manager, channel_manager.process_pending_events(&event_handler),
 				gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
-				channel_manager.get_persistable_update_future().wait_timeout(Duration::from_millis(100)),
+				Sleeper::from_two_futures(
+					channel_manager.get_persistable_update_future(),
+					chain_monitor.get_update_future()
+				).wait_timeout(Duration::from_millis(100)),
 				|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur)
 		});
 		Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }

lightning/src/chain/chainmonitor.rs

@@ -37,6 +37,7 @@ use crate::events::{Event, EventHandler};
 use crate::util::atomic_counter::AtomicCounter;
 use crate::util::logger::Logger;
 use crate::util::errors::APIError;
+use crate::util::wakers::{Future, Notifier};
 use crate::ln::channelmanager::ChannelDetails;
 
 use crate::prelude::*;
@@ -240,6 +241,8 @@ pub struct ChainMonitor<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T:
 	pending_monitor_events: Mutex<Vec<(OutPoint, Vec<MonitorEvent>, Option<PublicKey>)>>,
 	/// The best block height seen, used as a proxy for the passage of time.
 	highest_chain_height: AtomicUsize,
+
+	persistence_notifier: Notifier,
 }
 
 impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
@@ -300,6 +303,7 @@ where C::Target: chain::Filter,
 					ChannelMonitorUpdateStatus::PermanentFailure => {
 						monitor_state.channel_perm_failed.store(true, Ordering::Release);
 						self.pending_monitor_events.lock().unwrap().push((*funding_outpoint, vec![MonitorEvent::UpdateFailed(*funding_outpoint)], monitor.get_counterparty_node_id()));
+						self.persistence_notifier.notify();
 					},
 					ChannelMonitorUpdateStatus::InProgress => {
 						log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
@@ -345,6 +349,7 @@ where C::Target: chain::Filter,
 			persister,
 			pending_monitor_events: Mutex::new(Vec::new()),
 			highest_chain_height: AtomicUsize::new(0),
+			persistence_notifier: Notifier::new(),
 		}
 	}
 
@@ -472,6 +477,7 @@ where C::Target: chain::Filter,
 				}
 			},
 		}
+		self.persistence_notifier.notify();
 		Ok(())
 	}
 
@@ -486,6 +492,7 @@ where C::Target: chain::Filter,
 			funding_txo,
 			monitor_update_id,
 		}], counterparty_node_id));
+		self.persistence_notifier.notify();
 	}
 
 	#[cfg(any(test, fuzzing, feature = "_test_utils"))]
@@ -514,6 +521,18 @@ where C::Target: chain::Filter,
 			handler(event).await;
 		}
 	}
+
+	/// Gets a [`Future`] that completes when an event is available either via
+	/// [`chain::Watch::release_pending_monitor_events`] or
+	/// [`EventsProvider::process_pending_events`].
+	///
+	/// Note that callbacks registered on the [`Future`] MUST NOT call back into this
+	/// [`ChainMonitor`] and should instead register actions to be taken later.
+	///
+	/// [`EventsProvider::process_pending_events`]: crate::util::events::EventsProvider::process_pending_events
+	pub fn get_update_future(&self) -> Future {
+		self.persistence_notifier.get_future()
+	}
 }
 
 impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>