X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-block-sync%2Fsrc%2Fgossip.rs;h=9cd4049679c30fd5d7ddd2f86562d6df81758447;hb=4b70921c958181e43dc056dc05ef59427d13f2af;hp=3eb7ad4ae3537aae06c8e33edf59996614b4c740;hpb=01857b51a1e798ef344af845351b1c05c1c8a677;p=rust-lightning diff --git a/lightning-block-sync/src/gossip.rs b/lightning-block-sync/src/gossip.rs index 3eb7ad4a..9cd40496 100644 --- a/lightning-block-sync/src/gossip.rs +++ b/lightning-block-sync/src/gossip.rs @@ -2,24 +2,26 @@ //! 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}; use lightning::util::logger::Logger; -use std::sync::Arc; +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). @@ -27,9 +29,6 @@ use std::ops::Deref; /// Note that while this is implementable for a [`BlockSource`] which returns filtered block data /// (i.e. [`BlockData::HeaderOnly`] for [`BlockSource::get_block`] requests), such an /// implementation will reject all gossip as it is not fully able to verify the UTXOs referenced. -/// -/// For efficiency, an implementation may consider caching some set of blocks, as many redundant -/// calls may be made. pub trait UtxoSource : BlockSource + 'static { /// Fetches the block hash of the block at the given height. /// @@ -63,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), BlockSourceError>> + Unpin, + B: Future> + Unpin, +> { + pub a: A, + pub b: B, + a_res: Option<(BlockHash, Option)>, + b_res: Option, +} + +impl< + A: Future), BlockSourceError>> + Unpin, + B: Future> + Unpin, +> Joiner { + fn new(a: A, b: B) -> Self { Self { a, b, a_res: None, b_res: None } } +} + +impl< + A: Future), BlockSourceError>> + Unpin, + B: Future> + Unpin, +> Future for Joiner { + type Output = Result<((BlockHash, Option), BlockHash), BlockSourceError>; + fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll { + 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. /// @@ -74,74 +132,116 @@ impl FutureSpawner for TokioSpawner { pub struct GossipVerifier where Blocks::Target: UtxoSource, L::Target: Logger, - CM::Target: ChannelMessageHandler, - OM::Target: OnionMessageHandler, - CMH::Target: CustomMessageHandler, - NS::Target: NodeSigner, { source: Blocks, - peer_manager: Arc>, Self, L>>, OM, L, CMH, NS>>, + peer_manager_wake: Arc, gossiper: Arc>, Self, L>>, spawn: S, + block_cache: Arc>>, } +const BLOCK_CACHE_SIZE: usize = 5; + impl GossipVerifier where +> GossipVerifier where Blocks::Target: UtxoSource, L::Target: Logger, - CM::Target: ChannelMessageHandler, - OM::Target: OnionMessageHandler, - CMH::Target: CustomMessageHandler, - NS::Target: NodeSigner, { /// 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>, Self, L>>, peer_manager: Arc>, Self, L>>, OM, L, CMH, NS>>) -> Self { - Self { source, spawn, gossiper, peer_manager } + pub fn new( + source: Blocks, spawn: S, gossiper: Arc>, Self, L>>, peer_manager: APM + ) -> Self where APM::Target: APeerManager { + let peer_manager_wake = Arc::new(move || peer_manager.as_ref().process_events()); + Self { + source, spawn, gossiper, peer_manager_wake, + block_cache: Arc::new(Mutex::new(VecDeque::with_capacity(BLOCK_CACHE_SIZE))), + } } - async fn retrieve_utxo(source: Blocks, short_channel_id: u64) -> Result { + async fn retrieve_utxo( + source: Blocks, block_cache: Arc>>, short_channel_id: u64 + ) -> Result { let block_height = (short_channel_id >> 5 * 8) as u32; // block height is most significant three bytes let transaction_index = ((short_channel_id >> 2 * 8) & 0xffffff) as u32; let output_index = (short_channel_id & 0xffff) as u16; - let block_hash = source.get_block_hash_by_height(block_height).await - .map_err(|_| UtxoLookupError::UnknownTx)?; - let block_data = source.get_block(&block_hash).await - .map_err(|_| UtxoLookupError::UnknownTx)?; - let mut block = match block_data { - BlockData::HeaderOnly(_) => return Err(UtxoLookupError::UnknownTx), - BlockData::FullBlock(block) => block, + let (outpoint, output); + + 'tx_found: loop { // Used as a simple goto + macro_rules! process_block { + ($block: expr) => { { + if transaction_index as usize >= $block.txdata.len() { + return Err(UtxoLookupError::UnknownTx); + } + let transaction = &$block.txdata[transaction_index as usize]; + if output_index as usize >= transaction.output.len() { + return Err(UtxoLookupError::UnknownTx); + } + + outpoint = OutPoint::new(transaction.txid(), output_index.into()); + output = transaction.output[output_index as usize].clone(); + } } + } + { + let recent_blocks = block_cache.lock().unwrap(); + for (height, block) in recent_blocks.iter() { + if *height == block_height { + process_block!(block); + break 'tx_found; + } + } + } + + 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 { + BlockData::HeaderOnly(_) => return Err(UtxoLookupError::UnknownTx), + BlockData::FullBlock(block) => block, + }; + process_block!(block); + { + let mut recent_blocks = block_cache.lock().unwrap(); + let mut insert = true; + for (height, _) in recent_blocks.iter() { + if *height == block_height { + insert = false; + } + } + if insert { + if recent_blocks.len() >= BLOCK_CACHE_SIZE { + recent_blocks.pop_front(); + } + recent_blocks.push_back((block_height, block)); + } + } + break 'tx_found; }; - if transaction_index as usize >= block.txdata.len() { - return Err(UtxoLookupError::UnknownTx); - } - let mut transaction = block.txdata.swap_remove(transaction_index as usize); - if output_index as usize >= transaction.output.len() { - return Err(UtxoLookupError::UnknownTx); - } let outpoint_unspent = - source.is_output_unspent(OutPoint::new(transaction.txid(), output_index.into())).await - .map_err(|_| UtxoLookupError::UnknownTx)?; + source.is_output_unspent(outpoint).await.map_err(|_| UtxoLookupError::UnknownTx)?; if outpoint_unspent { - Ok(transaction.output.swap_remove(output_index as usize)) + Ok(output) } else { Err(UtxoLookupError::UnknownTx) } @@ -151,18 +251,9 @@ impl Deref for GossipVerifier where +> Deref for GossipVerifier where Blocks::Target: UtxoSource, L::Target: Logger, - CM::Target: ChannelMessageHandler, - OM::Target: OnionMessageHandler, - CMH::Target: CustomMessageHandler, - NS::Target: NodeSigner, { type Target = Self; fn deref(&self) -> &Self { self } @@ -172,29 +263,21 @@ impl UtxoLookup for GossipVerifier where +> UtxoLookup for GossipVerifier where Blocks::Target: UtxoSource, L::Target: Logger, - CM::Target: ChannelMessageHandler, - OM::Target: OnionMessageHandler, - CMH::Target: CustomMessageHandler, - NS::Target: NodeSigner, { - 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 pm = Arc::clone(&self.peer_manager); + 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, short_channel_id).await; + let res = Self::retrieve_utxo(source, block_cache, short_channel_id).await; fut.resolve(gossiper.network_graph(), &*gossiper, res); - pm.process_events(); + (pmw)(); }); UtxoResult::Async(res) }