X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-block-sync%2Fsrc%2Fgossip.rs;h=591b0298793ffb3baadb1d0ee0009b1199a95941;hb=1ee70af3cc4d261e698c3e3f6bfa0f8720867707;hp=4e66c0ce9756f016523bb9a9cfd2d65f973f5eca;hpb=3482fceeab251b2534bf0b7d0ff882db37409d40;p=rust-lightning

diff --git a/lightning-block-sync/src/gossip.rs b/lightning-block-sync/src/gossip.rs
index 4e66c0ce..591b0298 100644
--- a/lightning-block-sync/src/gossip.rs
+++ b/lightning-block-sync/src/gossip.rs
@@ -2,16 +2,14 @@
 //! current UTXO set. This module defines an implementation of the LDK API required to do so
 //! against a [`BlockSource`] which implements a few additional methods for accessing the UTXO set.
 
-use crate::{AsyncBlockSourceResult, BlockData, BlockSource};
+use crate::{AsyncBlockSourceResult, BlockData, BlockSource, BlockSourceError};
 
 use bitcoin::blockdata::block::Block;
+use bitcoin::blockdata::constants::ChainHash;
 use bitcoin::blockdata::transaction::{TxOut, OutPoint};
 use bitcoin::hash_types::BlockHash;
 
-use lightning::sign::NodeSigner;
-
-use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor};
-use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler};
+use lightning::ln::peer_handler::APeerManager;
 
 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 use lightning::routing::utxo::{UtxoFuture, UtxoLookup, UtxoResult, UtxoLookupError};
@@ -22,6 +20,8 @@ use std::sync::{Arc, Mutex};
 use std::collections::VecDeque;
 use std::future::Future;
 use std::ops::Deref;
+use std::pin::Pin;
+use std::task::Poll;
 
 /// A trait which extends [`BlockSource`] and can be queried to fetch the block at a given height
 /// as well as whether a given output is unspent (i.e. a member of the current UTXO set).
@@ -62,6 +62,65 @@ impl FutureSpawner for TokioSpawner {
 	}
 }
 
+/// A trivial future which joins two other futures and polls them at the same time, returning only
+/// once both complete.
+pub(crate) struct Joiner<
+	A: Future<Output=Result<(BlockHash, Option<u32>), BlockSourceError>> + Unpin,
+	B: Future<Output=Result<BlockHash, BlockSourceError>> + Unpin,
+> {
+	pub a: A,
+	pub b: B,
+	a_res: Option<(BlockHash, Option<u32>)>,
+	b_res: Option<BlockHash>,
+}
+
+impl<
+	A: Future<Output=Result<(BlockHash, Option<u32>), BlockSourceError>> + Unpin,
+	B: Future<Output=Result<BlockHash, BlockSourceError>> + Unpin,
+> Joiner<A, B> {
+	fn new(a: A, b: B) -> Self { Self { a, b, a_res: None, b_res: None } }
+}
+
+impl<
+	A: Future<Output=Result<(BlockHash, Option<u32>), BlockSourceError>> + Unpin,
+	B: Future<Output=Result<BlockHash, BlockSourceError>> + Unpin,
+> Future for Joiner<A, B> {
+	type Output = Result<((BlockHash, Option<u32>), BlockHash), BlockSourceError>;
+	fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<Self::Output> {
+		if self.a_res.is_none() {
+			match Pin::new(&mut self.a).poll(ctx) {
+				Poll::Ready(res) => {
+					if let Ok(ok) = res {
+						self.a_res = Some(ok);
+					} else {
+						return Poll::Ready(Err(res.unwrap_err()));
+					}
+				},
+				Poll::Pending => {},
+			}
+		}
+		if self.b_res.is_none() {
+			match Pin::new(&mut self.b).poll(ctx) {
+				Poll::Ready(res) => {
+					if let Ok(ok) = res {
+						self.b_res = Some(ok);
+					} else {
+						return Poll::Ready(Err(res.unwrap_err()));
+					}
+
+				},
+				Poll::Pending => {},
+			}
+		}
+		if let Some(b_res) = self.b_res {
+			if let Some(a_res) = self.a_res {
+				return Poll::Ready(Ok((a_res, b_res)))
+			}
+		}
+		Poll::Pending
+	}
+}
+
 /// A struct which wraps a [`UtxoSource`] and a few LDK objects and implements the LDK
 /// [`UtxoLookup`] trait.
 ///
@@ -73,21 +132,14 @@ impl FutureSpawner for TokioSpawner {
 pub struct GossipVerifier<S: FutureSpawner,
 	Blocks: Deref + Send + Sync + 'static + Clone,
 	L: Deref + Send + Sync + 'static,
-	Descriptor: SocketDescriptor + Send + Sync + 'static,
-	CM: Deref + Send + Sync + 'static,
-	OM: Deref + Send + Sync + 'static,
-	CMH: Deref + Send + Sync + 'static,
-	NS: Deref + Send + Sync + 'static,
+	APM: Deref + Send + Sync + 'static + Clone,
 > where
 	Blocks::Target: UtxoSource,
 	L::Target: Logger,
-	CM::Target: ChannelMessageHandler,
-	OM::Target: OnionMessageHandler,
-	CMH::Target: CustomMessageHandler,
-	NS::Target: NodeSigner,
+	APM::Target: APeerManager,
 {
 	source: Blocks,
-	peer_manager: Arc<PeerManager<Descriptor, CM, Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>, OM, L, CMH, NS>>,
+	peer_manager: APM,
 	gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>,
 	spawn: S,
 	block_cache: Arc<Mutex<VecDeque<(u32, Block)>>>,
@@ -98,24 +150,17 @@ const BLOCK_CACHE_SIZE: usize = 5;
 impl<S: FutureSpawner,
 	Blocks: Deref + Send + Sync + Clone,
 	L: Deref + Send + Sync,
-	Descriptor: SocketDescriptor + Send + Sync,
-	CM: Deref + Send + Sync,
-	OM: Deref + Send + Sync,
-	CMH: Deref + Send + Sync,
-	NS: Deref + Send + Sync,
-> GossipVerifier<S, Blocks, L, Descriptor, CM, OM, CMH, NS> where
+	APM: Deref + Send + Sync + Clone,
+> GossipVerifier<S, Blocks, L, APM> where
 	Blocks::Target: UtxoSource,
 	L::Target: Logger,
-	CM::Target: ChannelMessageHandler,
-	OM::Target: OnionMessageHandler,
-	CMH::Target: CustomMessageHandler,
-	NS::Target: NodeSigner,
+	APM::Target: APeerManager,
 {
 	/// Constructs a new [`GossipVerifier`].
 	///
 	/// 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(source: Blocks, spawn: S, gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>, peer_manager: Arc<PeerManager<Descriptor, CM, Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>, OM, L, CMH, NS>>) -> Self {
+	pub fn new(source: Blocks, spawn: S, gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>, peer_manager: APM) -> Self {
 		Self {
 			source, spawn, gossiper, peer_manager,
 			block_cache: Arc::new(Mutex::new(VecDeque::with_capacity(BLOCK_CACHE_SIZE))),
@@ -156,8 +201,20 @@ impl<S: FutureSpawner,
 				}
 			}
 
-			let block_hash = source.get_block_hash_by_height(block_height).await
+			let ((_, tip_height_opt), block_hash) =
+				Joiner::new(source.get_best_block(), source.get_block_hash_by_height(block_height))
+				.await
 				.map_err(|_| UtxoLookupError::UnknownTx)?;
+			if let Some(tip_height) = tip_height_opt {
+				// If the block doesn't yet have five confirmations, error out.
+				//
+				// The BOLT spec requires nodes wait for six confirmations before announcing a
+				// channel, and we give them one block of headroom in case we're delayed seeing a
+				// block.
+				if block_height + 5 > tip_height {
+					return Err(UtxoLookupError::UnknownTx);
+				}
+			}
 			let block_data = source.get_block(&block_hash).await
 				.map_err(|_| UtxoLookupError::UnknownTx)?;
 			let block = match block_data {
@@ -195,18 +252,11 @@ impl<S: FutureSpawner,
 impl<S: FutureSpawner,
 	Blocks: Deref + Send + Sync + Clone,
 	L: Deref + Send + Sync,
-	Descriptor: SocketDescriptor + Send + Sync,
-	CM: Deref + Send + Sync,
-	OM: Deref + Send + Sync,
-	CMH: Deref + Send + Sync,
-	NS: Deref + Send + Sync,
-> Deref for GossipVerifier<S, Blocks, L, Descriptor, CM, OM, CMH, NS> where
+	APM: Deref + Send + Sync + Clone,
+> Deref for GossipVerifier<S, Blocks, L, APM> where
 	Blocks::Target: UtxoSource,
 	L::Target: Logger,
-	CM::Target: ChannelMessageHandler,
-	OM::Target: OnionMessageHandler,
-	CMH::Target: CustomMessageHandler,
-	NS::Target: NodeSigner,
+	APM::Target: APeerManager,
 {
 	type Target = Self;
 	fn deref(&self) -> &Self { self }
@@ -216,30 +266,23 @@ impl<S: FutureSpawner,
 impl<S: FutureSpawner,
 	Blocks: Deref + Send + Sync + Clone,
 	L: Deref + Send + Sync,
-	Descriptor: SocketDescriptor + Send + Sync,
-	CM: Deref + Send + Sync,
-	OM: Deref + Send + Sync,
-	CMH: Deref + Send + Sync,
-	NS: Deref + Send + Sync,
-> UtxoLookup for GossipVerifier<S, Blocks, L, Descriptor, CM, OM, CMH, NS> where
+	APM: Deref + Send + Sync + Clone,
+> UtxoLookup for GossipVerifier<S, Blocks, L, APM> where
 	Blocks::Target: UtxoSource,
 	L::Target: Logger,
-	CM::Target: ChannelMessageHandler,
-	OM::Target: OnionMessageHandler,
-	CMH::Target: CustomMessageHandler,
-	NS::Target: NodeSigner,
+	APM::Target: APeerManager,
 {
-	fn get_utxo(&self, _genesis_hash: &BlockHash, short_channel_id: u64) -> UtxoResult {
+	fn get_utxo(&self, _chain_hash: &ChainHash, short_channel_id: u64) -> UtxoResult {
 		let res = UtxoFuture::new();
 		let fut = res.clone();
 		let source = self.source.clone();
 		let gossiper = Arc::clone(&self.gossiper);
 		let block_cache = Arc::clone(&self.block_cache);
-		let pm = Arc::clone(&self.peer_manager);
+		let pm = self.peer_manager.clone();
 		self.spawn.spawn(async move {
 			let res = Self::retrieve_utxo(source, block_cache, short_channel_id).await;
 			fut.resolve(gossiper.network_graph(), &*gossiper, res);
-			pm.process_events();
+			pm.as_ref().process_events();
 		});
 		UtxoResult::Async(res)
 	}