Move to awaiting gossip validation in the background processor

`P2PGossipSync` is a rather poor design. It currently basically
requires two circular `Arc` references, leaving `NetworkGraph`s to
leak if LDK is un-loaded:
 * `P2PGossipSync` owns/holds a reference to the
   `GossipVerifier` and `GossipVerifier` holds an `Arc` to the
   `P2PGossipSync` and
 * `PeerManager` holds a reference to the `P2PGossipSync` (as the
   gossip message handler) which owns/holds a reference to the
   `GossipVerifier`, which has a `Deref` (likely an `Arc` in
   practice) to the `PeerManager`.

Instead, we should move towards the same design we have elsewhere -
hold a `Notifier` and expose waiting on it to the background
processor then poll for completion from there (in this case, as in
others by checking for completion when handling
`get_and_clear_pending_msg_events` calls).

After the last few commits of setup, here we finally switch to
waking the background processor directly when we detect async
gossip validation completion, allowing us to drop the circular
references in `P2PGossipSync`/`GossipVerifier` entirely.

Fixes #3369

Author: TheBlueMatt

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() {

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,35 +235,28 @@ 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 {
 		self
 	}
 }
 
-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)
 	}

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;