Move to awaiting gossip validation in the background processor

diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index 36b563f..7361d02 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -64,6 +64,7 @@ use lightning::util::persist::{
 	SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::sweep::{OutputSweeper, OutputSweeperSync};
+use lightning::util::wakers::Future;
 #[cfg(feature = "std")]
 use lightning::util::wakers::Sleeper;
 use lightning_rapid_gossip_sync::RapidGossipSync;
@@ -235,6 +236,14 @@ where
 			GossipSync::None => None,
 		}
 	}
+
+	fn validation_completion_future(&self) -> Option<Future> {
+		match self {
+			GossipSync::P2P(gossip_sync) => Some(gossip_sync.validation_completion_future()),
+			GossipSync::Rapid(_) => None,
+			GossipSync::None => None,
+		}
+	}
 }
 
 /// This is not exported to bindings users as the bindings concretize everything and have constructors for us
@@ -520,12 +529,14 @@ pub(crate) mod futures_util {
 		C: Future<Output = ()> + Unpin,
 		D: Future<Output = ()> + Unpin,
 		E: Future<Output = ()> + Unpin,
+		F: Future<Output = ()> + Unpin,
 	> {
 		pub a: A,
 		pub b: B,
 		pub c: C,
 		pub d: D,
 		pub e: E,
+		pub f: F,
 	}
 
 	pub(crate) enum SelectorOutput {
@@ -534,6 +545,7 @@ pub(crate) mod futures_util {
 		C,
 		D,
 		E,
+		F,
 	}
 
 	impl<
@@ -542,7 +554,8 @@ pub(crate) mod futures_util {
 			C: Future<Output = ()> + Unpin,
 			D: Future<Output = ()> + Unpin,
 			E: Future<Output = ()> + Unpin,
-		> Future for Selector<A, B, C, D, E>
+			F: Future<Output = ()> + Unpin,
+		> Future for Selector<A, B, C, D, E, F>
 	{
 		type Output = SelectorOutput;
 		fn poll(
@@ -580,6 +593,12 @@ pub(crate) mod futures_util {
 				},
 				Poll::Pending => {},
 			}
+			match Pin::new(&mut self.f).poll(ctx) {
+				Poll::Ready(()) => {
+					return Poll::Ready(SelectorOutput::F);
+				},
+				Poll::Pending => {},
+			}
 			Poll::Pending
 		}
 	}
@@ -606,6 +625,12 @@ pub(crate) mod futures_util {
 		}
 	}
 
+	impl<F: Future<Output = ()> + Unpin> From<Option<F>> for OptionalSelector<F> {
+		fn from(optional_future: Option<F>) -> Self {
+			Self { optional_future }
+		}
+	}
+
 	// If we want to poll a future without an async context to figure out if it has completed or
 	// not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
 	// but sadly there's a good bit of boilerplate here.
@@ -1058,18 +1083,13 @@ where
 		if mobile_interruptable_platform {
 			await_start = Some(sleeper(Duration::from_secs(1)));
 		}
-		let om_fut = if let Some(om) = onion_messenger.as_ref() {
-			let fut = om.get_om().get_update_future();
-			OptionalSelector { optional_future: Some(fut) }
-		} else {
-			OptionalSelector { optional_future: None }
-		};
-		let lm_fut = if let Some(lm) = liquidity_manager.as_ref() {
-			let fut = lm.get_lm().get_pending_msgs_or_needs_persist_future();
-			OptionalSelector { optional_future: Some(fut) }
-		} else {
-			OptionalSelector { optional_future: None }
-		};
+		let om_fut: OptionalSelector<_> =
+			onion_messenger.as_ref().map(|om| om.get_om().get_update_future()).into();
+		let lm_fut: OptionalSelector<_> = liquidity_manager
+			.as_ref()
+			.map(|lm| lm.get_lm().get_pending_msgs_or_needs_persist_future())
+			.into();
+		let gv_fut: OptionalSelector<_> = gossip_sync.validation_completion_future().into();
 		let needs_processing = channel_manager.get_cm().needs_pending_htlc_processing();
 		let sleep_delay = match (needs_processing, mobile_interruptable_platform) {
 			(true, true) => batch_delay.get().min(Duration::from_millis(100)),
@@ -1083,9 +1103,14 @@ where
 			c: chain_monitor.get_update_future(),
 			d: om_fut,
 			e: lm_fut,
+			f: gv_fut,
 		};
 		match fut.await {
-			SelectorOutput::B | SelectorOutput::C | SelectorOutput::D | SelectorOutput::E => {},
+			SelectorOutput::B
+			| SelectorOutput::C
+			| SelectorOutput::D
+			| SelectorOutput::E
+			| SelectorOutput::F => {},
 			SelectorOutput::A(exit) => {
 				if exit {
 					break;
@@ -1639,11 +1664,12 @@ impl BackgroundProcessor {
 				let lm_fut = liquidity_manager
 					.as_ref()
 					.map(|lm| lm.get_lm().get_pending_msgs_or_needs_persist_future());
+				let gv_fut = gossip_sync.validation_completion_future();
 				let always_futures = [
 					channel_manager.get_cm().get_event_or_persistence_needed_future(),
 					chain_monitor.get_update_future(),
 				];
-				let futures = always_futures.into_iter().chain(om_fut).chain(lm_fut);
+				let futures = always_futures.into_iter().chain(om_fut).chain(lm_fut).chain(gv_fut);
 				let sleeper = Sleeper::from_futures(futures);
 
 				let batch_delay = if channel_manager.get_cm().needs_pending_htlc_processing() {
diff --git a/lightning-block-sync/src/gossip.rs b/lightning-block-sync/src/gossip.rs
index 63045b6..00d3216 100644
--- a/lightning-block-sync/src/gossip.rs
+++ b/lightning-block-sync/src/gossip.rs
@@ -9,10 +9,7 @@ use bitcoin::constants::ChainHash;
 use bitcoin::hash_types::BlockHash;
 use bitcoin::transaction::{OutPoint, TxOut};
 
-use lightning::ln::peer_handler::APeerManager;
-use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 use lightning::routing::utxo::{UtxoFuture, UtxoLookup, UtxoLookupError, UtxoResult};
-use lightning::util::logger::Logger;
 use lightning::util::native_async::FutureSpawner;
 use lightning::util::wakers::Notifier;
 
@@ -128,46 +125,28 @@ impl<
 /// value of 1024 should more than suffice), and ensure you have sufficient file descriptors
 /// available on both Bitcoin Core and your LDK application for each request to hold its own
 /// connection.
-pub struct GossipVerifier<
-	S: FutureSpawner,
-	Blocks: Deref + Send + Sync + 'static + Clone,
-	L: Deref + Send + Sync + 'static,
-> where
+pub struct GossipVerifier<S: FutureSpawner, Blocks: Deref + Send + Sync + 'static + Clone>
+where
 	Blocks::Target: UtxoSource,
-	L::Target: Logger,
 {
 	source: Blocks,
-	peer_manager_wake: Arc<dyn Fn() + Send + Sync>,
-	gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Arc<Self>, L>>,
 	spawn: S,
 	block_cache: Arc<Mutex<VecDeque<(u32, Block)>>>,
 }
 
 const BLOCK_CACHE_SIZE: usize = 5;
 
-impl<S: FutureSpawner, Blocks: Deref + Send + Sync + Clone, L: Deref + Send + Sync>
-	GossipVerifier<S, Blocks, L>
+impl<S: FutureSpawner, Blocks: Deref + Send + Sync + Clone> GossipVerifier<S, Blocks>
 where
 	Blocks::Target: UtxoSource,
-	L::Target: Logger,
 {
-	/// Constructs a new [`GossipVerifier`].
+	/// Constructs a new [`GossipVerifier`] for use in a [`P2PGossipSync`].
 	///
-	/// This is expected to be given to a [`P2PGossipSync`] (initially constructed with `None` for
-	/// the UTXO lookup) via [`P2PGossipSync::add_utxo_lookup`].
-	pub fn new<APM: Deref + Send + Sync + Clone + 'static>(
-		source: Blocks, spawn: S, gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Arc<Self>, L>>,
-		peer_manager: APM,
-	) -> Self
-	where
-		APM::Target: APeerManager,
-	{
-		let peer_manager_wake = Arc::new(move || peer_manager.as_ref().process_events());
+	/// [`P2PGossipSync`]: lightning::routing::gossip::P2PGossipSync
+	pub fn new(source: Blocks, spawn: S) -> Self {
 		Self {
 			source,
 			spawn,
-			gossiper,
-			peer_manager_wake,
 			block_cache: Arc::new(Mutex::new(VecDeque::with_capacity(BLOCK_CACHE_SIZE))),
 		}
 	}
@@ -256,11 +235,9 @@ where
 	}
 }
 
-impl<S: FutureSpawner, Blocks: Deref + Send + Sync + Clone, L: Deref + Send + Sync> Deref
-	for GossipVerifier<S, Blocks, L>
+impl<S: FutureSpawner, Blocks: Deref + Send + Sync + Clone> Deref for GossipVerifier<S, Blocks>
 where
 	Blocks::Target: UtxoSource,
-	L::Target: Logger,
 {
 	type Target = Self;
 	fn deref(&self) -> &Self {
@@ -268,23 +245,18 @@ where
 	}
 }
 
-impl<S: FutureSpawner, Blocks: Deref + Send + Sync + Clone, L: Deref + Send + Sync> UtxoLookup
-	for GossipVerifier<S, Blocks, L>
+impl<S: FutureSpawner, Blocks: Deref + Send + Sync + Clone> UtxoLookup for GossipVerifier<S, Blocks>
 where
 	Blocks::Target: UtxoSource,
-	L::Target: Logger,
 {
 	fn get_utxo(&self, _chain_hash: &ChainHash, scid: u64, notifier: Arc<Notifier>) -> UtxoResult {
 		let res = UtxoFuture::new(notifier);
 		let fut = res.clone();
 		let source = self.source.clone();
-		let gossiper = Arc::clone(&self.gossiper);
 		let block_cache = Arc::clone(&self.block_cache);
-		let pmw = Arc::clone(&self.peer_manager_wake);
 		self.spawn.spawn(async move {
 			let res = Self::retrieve_utxo(source, block_cache, scid).await;
 			fut.resolve(res);
-			(pmw)();
 		});
 		UtxoResult::Async(res)
 	}
diff --git a/lightning/src/routing/gossip.rs b/lightning/src/routing/gossip.rs
index 46ca332..e8fcb7b 100644
--- a/lightning/src/routing/gossip.rs
+++ b/lightning/src/routing/gossip.rs
@@ -43,6 +43,7 @@ use crate::util::indexed_map::{
 use crate::util::logger::{Level, Logger};
 use crate::util::scid_utils::{block_from_scid, scid_from_parts, MAX_SCID_BLOCK};
 use crate::util::ser::{MaybeReadable, Readable, ReadableArgs, RequiredWrapper, Writeable, Writer};
+use crate::util::wakers::Future;
 
 use crate::io;
 use crate::io_extras::{copy, sink};
@@ -367,6 +368,17 @@ where
 		&self.network_graph
 	}
 
+	/// Gets a [`Future`] which will resolve the next time an async validation of gossip data
+	/// completes.
+	///
+	/// If the [`UtxoLookup`] provided in [`P2PGossipSync::new`] does not return
+	/// [`UtxoResult::Async`] values, the returned [`Future`] will never resolve
+	///
+	/// [`UtxoResult::Async`]: crate::routing::utxo::UtxoResult::Async
+	pub fn validation_completion_future(&self) -> Future {
+		self.network_graph.pending_checks.completion_notifier.get_future()
+	}
+
 	/// Returns true when a full routing table sync should be performed with a peer.
 	fn should_request_full_sync(&self) -> bool {
 		const FULL_SYNCS_TO_REQUEST: usize = 5;