Have background processor task drive `LiquidityManger` persistence

diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index ef6e82a..44ce52b 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -571,31 +571,34 @@ pub(crate) mod futures_util {
 		unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) }
 	}
 
-	enum JoinerResult<E, F: Future<Output = Result<(), E>> + Unpin> {
+	enum JoinerResult<ERR, F: Future<Output = Result<(), ERR>> + Unpin> {
 		Pending(Option<F>),
-		Ready(Result<(), E>),
+		Ready(Result<(), ERR>),
 	}
 
 	pub(crate) struct Joiner<
-		E,
-		A: Future<Output = Result<(), E>> + Unpin,
-		B: Future<Output = Result<(), E>> + Unpin,
-		C: Future<Output = Result<(), E>> + Unpin,
-		D: Future<Output = Result<(), E>> + Unpin,
+		ERR,
+		A: Future<Output = Result<(), ERR>> + Unpin,
+		B: Future<Output = Result<(), ERR>> + Unpin,
+		C: Future<Output = Result<(), ERR>> + Unpin,
+		D: Future<Output = Result<(), ERR>> + Unpin,
+		E: Future<Output = Result<(), ERR>> + Unpin,
 	> {
-		a: JoinerResult<E, A>,
-		b: JoinerResult<E, B>,
-		c: JoinerResult<E, C>,
-		d: JoinerResult<E, D>,
+		a: JoinerResult<ERR, A>,
+		b: JoinerResult<ERR, B>,
+		c: JoinerResult<ERR, C>,
+		d: JoinerResult<ERR, D>,
+		e: JoinerResult<ERR, E>,
 	}
 
 	impl<
-			E,
-			A: Future<Output = Result<(), E>> + Unpin,
-			B: Future<Output = Result<(), E>> + Unpin,
-			C: Future<Output = Result<(), E>> + Unpin,
-			D: Future<Output = Result<(), E>> + Unpin,
-		> Joiner<E, A, B, C, D>
+			ERR,
+			A: Future<Output = Result<(), ERR>> + Unpin,
+			B: Future<Output = Result<(), ERR>> + Unpin,
+			C: Future<Output = Result<(), ERR>> + Unpin,
+			D: Future<Output = Result<(), ERR>> + Unpin,
+			E: Future<Output = Result<(), ERR>> + Unpin,
+		> Joiner<ERR, A, B, C, D, E>
 	{
 		pub(crate) fn new() -> Self {
 			Self {
@@ -603,13 +606,14 @@ pub(crate) mod futures_util {
 				b: JoinerResult::Pending(None),
 				c: JoinerResult::Pending(None),
 				d: JoinerResult::Pending(None),
+				e: JoinerResult::Pending(None),
 			}
 		}
 
 		pub(crate) fn set_a(&mut self, fut: A) {
 			self.a = JoinerResult::Pending(Some(fut));
 		}
-		pub(crate) fn set_a_res(&mut self, res: Result<(), E>) {
+		pub(crate) fn set_a_res(&mut self, res: Result<(), ERR>) {
 			self.a = JoinerResult::Ready(res);
 		}
 		pub(crate) fn set_b(&mut self, fut: B) {
@@ -621,19 +625,23 @@ pub(crate) mod futures_util {
 		pub(crate) fn set_d(&mut self, fut: D) {
 			self.d = JoinerResult::Pending(Some(fut));
 		}
+		pub(crate) fn set_e(&mut self, fut: E) {
+			self.e = JoinerResult::Pending(Some(fut));
+		}
 	}
 
 	impl<
-			E,
-			A: Future<Output = Result<(), E>> + Unpin,
-			B: Future<Output = Result<(), E>> + Unpin,
-			C: Future<Output = Result<(), E>> + Unpin,
-			D: Future<Output = Result<(), E>> + Unpin,
-		> Future for Joiner<E, A, B, C, D>
+			ERR,
+			A: Future<Output = Result<(), ERR>> + Unpin,
+			B: Future<Output = Result<(), ERR>> + Unpin,
+			C: Future<Output = Result<(), ERR>> + Unpin,
+			D: Future<Output = Result<(), ERR>> + Unpin,
+			E: Future<Output = Result<(), ERR>> + Unpin,
+		> Future for Joiner<ERR, A, B, C, D, E>
 	where
-		Joiner<E, A, B, C, D>: Unpin,
+		Joiner<ERR, A, B, C, D, E>: Unpin,
 	{
-		type Output = [Result<(), E>; 4];
+		type Output = [Result<(), ERR>; 5];
 		fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<Self::Output> {
 			let mut all_complete = true;
 			macro_rules! handle {
@@ -642,7 +650,7 @@ pub(crate) mod futures_util {
 						JoinerResult::Pending(None) => {
 							self.$val = JoinerResult::Ready(Ok(()));
 						},
-						JoinerResult::<E, _>::Pending(Some(ref mut val)) => {
+						JoinerResult::<ERR, _>::Pending(Some(ref mut val)) => {
 							match Pin::new(val).poll(ctx) {
 								Poll::Ready(res) => {
 									self.$val = JoinerResult::Ready(res);
@@ -660,9 +668,10 @@ pub(crate) mod futures_util {
 			handle!(b);
 			handle!(c);
 			handle!(d);
+			handle!(e);
 
 			if all_complete {
-				let mut res = [Ok(()), Ok(()), Ok(()), Ok(())];
+				let mut res = [Ok(()), Ok(()), Ok(()), Ok(()), Ok(())];
 				if let JoinerResult::Ready(ref mut val) = &mut self.a {
 					core::mem::swap(&mut res[0], val);
 				}
@@ -675,6 +684,9 @@ pub(crate) mod futures_util {
 				if let JoinerResult::Ready(ref mut val) = &mut self.d {
 					core::mem::swap(&mut res[3], val);
 				}
+				if let JoinerResult::Ready(ref mut val) = &mut self.e {
+					core::mem::swap(&mut res[4], val);
+				}
 				Poll::Ready(res)
 			} else {
 				Poll::Pending
@@ -1003,7 +1015,7 @@ where
 			OptionalSelector { optional_future: None }
 		};
 		let lm_fut = if let Some(lm) = liquidity_manager.as_ref() {
-			let fut = lm.get_lm().get_pending_msgs_future();
+			let fut = lm.get_lm().get_pending_msgs_or_needs_persist_future();
 			OptionalSelector { optional_future: Some(fut) }
 		} else {
 			OptionalSelector { optional_future: None }
@@ -1206,6 +1218,17 @@ where
 			None => {},
 		}
 
+		if let Some(liquidity_manager) = liquidity_manager.as_ref() {
+			log_trace!(logger, "Persisting LiquidityManager...");
+			let fut = async {
+				liquidity_manager.get_lm().persist().await.map_err(|e| {
+					log_error!(logger, "Persisting LiquidityManager failed: {}", e);
+					e
+				})
+			};
+			futures.set_e(Box::pin(fut));
+		}
+
 		// Run persistence tasks in parallel and exit if any of them returns an error.
 		for res in futures.await {
 			res?;
@@ -1562,7 +1585,7 @@ impl BackgroundProcessor {
 						&channel_manager.get_cm().get_event_or_persistence_needed_future(),
 						&chain_monitor.get_update_future(),
 						&om.get_om().get_update_future(),
-						&lm.get_lm().get_pending_msgs_future(),
+						&lm.get_lm().get_pending_msgs_or_needs_persist_future(),
 					),
 					(Some(om), None) => Sleeper::from_three_futures(
 						&channel_manager.get_cm().get_event_or_persistence_needed_future(),
@@ -1572,7 +1595,7 @@ impl BackgroundProcessor {
 					(None, Some(lm)) => Sleeper::from_three_futures(
 						&channel_manager.get_cm().get_event_or_persistence_needed_future(),
 						&chain_monitor.get_update_future(),
-						&lm.get_lm().get_pending_msgs_future(),
+						&lm.get_lm().get_pending_msgs_or_needs_persist_future(),
 					),
 					(None, None) => Sleeper::from_two_futures(
 						&channel_manager.get_cm().get_event_or_persistence_needed_future(),
@@ -1606,6 +1629,13 @@ impl BackgroundProcessor {
 					log_trace!(logger, "Done persisting ChannelManager.");
 				}
 
+				if let Some(liquidity_manager) = liquidity_manager.as_ref() {
+					log_trace!(logger, "Persisting LiquidityManager...");
+					let _ = liquidity_manager.get_lm().persist().map_err(|e| {
+						log_error!(logger, "Persisting LiquidityManager failed: {}", e);
+					});
+				}
+
 				// Note that we want to run a graph prune once not long after startup before
 				// falling back to our usual hourly prunes. This avoids short-lived clients never
 				// pruning their network graph. We run once 60 seconds after startup before
diff --git a/lightning-liquidity/src/lsps0/client.rs b/lightning-liquidity/src/lsps0/client.rs
index 56dfb24..9c019aa 100644
--- a/lightning-liquidity/src/lsps0/client.rs
+++ b/lightning-liquidity/src/lsps0/client.rs
@@ -123,6 +123,7 @@ mod tests {
 
 	use lightning::util::persist::KVStoreSyncWrapper;
 	use lightning::util::test_utils::TestStore;
+	use lightning::util::wakers::Notifier;
 
 	use crate::lsps0::ser::{LSPSMessage, LSPSRequestId};
 	use crate::tests::utils::{self, TestEntropy};
@@ -131,7 +132,8 @@ mod tests {
 
 	#[test]
 	fn test_list_protocols() {
-		let pending_messages = Arc::new(MessageQueue::new());
+		let notifier = Arc::new(Notifier::new());
+		let pending_messages = Arc::new(MessageQueue::new(notifier));
 		let entropy_source = Arc::new(TestEntropy {});
 		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
 		let event_queue = Arc::new(EventQueue::new(VecDeque::new(), kv_store));
diff --git a/lightning-liquidity/src/lsps0/service.rs b/lightning-liquidity/src/lsps0/service.rs
index 2b4e678..e71150c 100644
--- a/lightning-liquidity/src/lsps0/service.rs
+++ b/lightning-liquidity/src/lsps0/service.rs
@@ -87,13 +87,15 @@ mod tests {
 	use crate::tests::utils;
 	use alloc::string::ToString;
 	use alloc::sync::Arc;
+	use lightning::util::wakers::Notifier;
 
 	use super::*;
 
 	#[test]
 	fn test_handle_list_protocols_request() {
 		let protocols: Vec<u16> = vec![];
-		let pending_messages = Arc::new(MessageQueue::new());
+		let notifier = Arc::new(Notifier::new());
+		let pending_messages = Arc::new(MessageQueue::new(notifier));
 
 		let lsps0_handler =
 			Arc::new(LSPS0ServiceHandler::new(protocols, Arc::clone(&pending_messages)));
diff --git a/lightning-liquidity/src/lsps5/client.rs b/lightning-liquidity/src/lsps5/client.rs
index 1045c1b..a643238 100644
--- a/lightning-liquidity/src/lsps5/client.rs
+++ b/lightning-liquidity/src/lsps5/client.rs
@@ -450,6 +450,7 @@ mod tests {
 	use core::sync::atomic::{AtomicU64, Ordering};
 	use lightning::util::persist::KVStoreSyncWrapper;
 	use lightning::util::test_utils::TestStore;
+	use lightning::util::wakers::Notifier;
 
 	struct UniqueTestEntropy {
 		counter: AtomicU64,
@@ -472,7 +473,8 @@ mod tests {
 		PublicKey,
 	) {
 		let test_entropy_source = Arc::new(UniqueTestEntropy { counter: AtomicU64::new(2) });
-		let message_queue = Arc::new(MessageQueue::new());
+		let notifier = Arc::new(Notifier::new());
+		let message_queue = Arc::new(MessageQueue::new(notifier));
 
 		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
 		let event_queue = Arc::new(EventQueue::new(VecDeque::new(), kv_store));
diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs
index 27e82f5..3ba1638 100644
--- a/lightning-liquidity/src/manager.rs
+++ b/lightning-liquidity/src/manager.rs
@@ -52,7 +52,7 @@ use lightning::sign::{EntropySource, NodeSigner};
 use lightning::util::logger::Level;
 use lightning::util::persist::{KVStore, KVStoreSync, KVStoreSyncWrapper};
 use lightning::util::ser::{LengthLimitedRead, LengthReadable};
-use lightning::util::wakers::Future;
+use lightning::util::wakers::{Future, Notifier};
 
 use lightning_types::features::{InitFeatures, NodeFeatures};
 
@@ -312,6 +312,7 @@ pub struct LiquidityManager<
 	_client_config: Option<LiquidityClientConfig>,
 	best_block: RwLock<Option<BestBlock>>,
 	_chain_source: Option<C>,
+	pending_msgs_or_needs_persist_notifier: Arc<Notifier>,
 }
 
 #[cfg(feature = "time")]
@@ -384,7 +385,9 @@ where
 		service_config: Option<LiquidityServiceConfig>,
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
 	) -> Result<Self, lightning::io::Error> {
-		let pending_messages = Arc::new(MessageQueue::new());
+		let pending_msgs_or_needs_persist_notifier = Arc::new(Notifier::new());
+		let pending_messages =
+			Arc::new(MessageQueue::new(Arc::clone(&pending_msgs_or_needs_persist_notifier)));
 		let persisted_queue = read_event_queue(kv_store.clone()).await?.unwrap_or_default();
 		let pending_events = Arc::new(EventQueue::new(persisted_queue, kv_store.clone()));
 		let ignored_peers = RwLock::new(new_hash_set());
@@ -523,6 +526,7 @@ where
 			_client_config: client_config,
 			best_block: RwLock::new(chain_params.map(|chain_params| chain_params.best_block)),
 			_chain_source: chain_source,
+			pending_msgs_or_needs_persist_notifier,
 		})
 	}
 
@@ -581,12 +585,12 @@ where
 	}
 
 	/// Returns a [`Future`] that will complete when the next batch of pending messages is ready to
-	/// be processed.
+	/// be processed *or* we need to be repersisted.
 	///
 	/// Note that callbacks registered on the [`Future`] MUST NOT call back into this
 	/// [`LiquidityManager`] and should instead register actions to be taken later.
-	pub fn get_pending_msgs_future(&self) -> Future {
-		self.pending_messages.get_pending_msgs_future()
+	pub fn get_pending_msgs_or_needs_persist_future(&self) -> Future {
+		self.pending_msgs_or_needs_persist_notifier.get_future()
 	}
 
 	/// Blocks the current thread until next event is ready and returns it.
@@ -1208,11 +1212,11 @@ where
 	}
 
 	/// Returns a [`Future`] that will complete when the next batch of pending messages is ready to
-	/// be processed.
+	/// be processed *or* we need to be repersisted.
 	///
-	/// Wraps [`LiquidityManager::get_pending_msgs_future`].
-	pub fn get_pending_msgs_future(&self) -> Future {
-		self.inner.get_pending_msgs_future()
+	/// Wraps [`LiquidityManager::get_pending_msgs_or_needs_persist_future`].
+	pub fn get_pending_msgs_or_needs_persist_future(&self) -> Future {
+		self.inner.get_pending_msgs_or_needs_persist_future()
 	}
 
 	/// Blocks the current thread until next event is ready and returns it.
diff --git a/lightning-liquidity/src/message_queue.rs b/lightning-liquidity/src/message_queue.rs
index d097573..8b248d8 100644
--- a/lightning-liquidity/src/message_queue.rs
+++ b/lightning-liquidity/src/message_queue.rs
@@ -13,9 +13,9 @@ use alloc::collections::VecDeque;
 use alloc::vec::Vec;
 
 use crate::lsps0::ser::LSPSMessage;
-use crate::sync::Mutex;
+use crate::sync::{Arc, Mutex};
 
-use lightning::util::wakers::{Future, Notifier};
+use lightning::util::wakers::Notifier;
 
 use bitcoin::secp256k1::PublicKey;
 
@@ -24,13 +24,12 @@ use bitcoin::secp256k1::PublicKey;
 /// [`LiquidityManager`]: crate::LiquidityManager
 pub struct MessageQueue {
 	queue: Mutex<VecDeque<(PublicKey, LSPSMessage)>>,
-	pending_msgs_notifier: Notifier,
+	pending_msgs_notifier: Arc<Notifier>,
 }
 
 impl MessageQueue {
-	pub(crate) fn new() -> Self {
+	pub(crate) fn new(pending_msgs_notifier: Arc<Notifier>) -> Self {
 		let queue = Mutex::new(VecDeque::new());
-		let pending_msgs_notifier = Notifier::new();
 		Self { queue, pending_msgs_notifier }
 	}
 
@@ -38,10 +37,6 @@ impl MessageQueue {
 		self.queue.lock().unwrap().drain(..).collect()
 	}
 
-	pub(crate) fn get_pending_msgs_future(&self) -> Future {
-		self.pending_msgs_notifier.get_future()
-	}
-
 	pub(crate) fn notifier(&self) -> MessageQueueNotifierGuard<'_> {
 		MessageQueueNotifierGuard { msg_queue: self, buffer: VecDeque::new() }
 	}