_ Git - rust-lightning/blob - lightning-block-sync/src/gossip.rs

   1 //! When fetching gossip from peers, lightning nodes need to validate that gossip against the
   2 //! current UTXO set. This module defines an implementation of the LDK API required to do so
   3 //! against a [`BlockSource`] which implements a few additional methods for accessing the UTXO set.
   4
   5 use crate::{AsyncBlockSourceResult, BlockData, BlockSource, BlockSourceError};
   6
   7 use bitcoin::blockdata::block::Block;
   8 use bitcoin::blockdata::constants::ChainHash;
   9 use bitcoin::blockdata::transaction::{TxOut, OutPoint};
  10 use bitcoin::hash_types::BlockHash;
  11
  12 use lightning::ln::peer_handler::APeerManager;
  13
  14 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
  15 use lightning::routing::utxo::{UtxoFuture, UtxoLookup, UtxoResult, UtxoLookupError};
  16
  17 use lightning::util::logger::Logger;
  18
  19 use std::sync::{Arc, Mutex};
  20 use std::collections::VecDeque;
  21 use std::future::Future;
  22 use std::ops::Deref;
  23 use std::pin::Pin;
  24 use std::task::Poll;
  25
  26 /// A trait which extends [`BlockSource`] and can be queried to fetch the block at a given height
  27 /// as well as whether a given output is unspent (i.e. a member of the current UTXO set).
  28 ///
  29 /// Note that while this is implementable for a [`BlockSource`] which returns filtered block data
  30 /// (i.e. [`BlockData::HeaderOnly`] for [`BlockSource::get_block`] requests), such an
  31 /// implementation will reject all gossip as it is not fully able to verify the UTXOs referenced.
  32 pub trait UtxoSource : BlockSource + 'static {
  33         /// Fetches the block hash of the block at the given height.
  34         ///
  35         /// This will, in turn, be passed to to [`BlockSource::get_block`] to fetch the block needed
  36         /// for gossip validation.
  37         fn get_block_hash_by_height<'a>(&'a self, block_height: u32) -> AsyncBlockSourceResult<'a, BlockHash>;
  38
  39         /// Returns true if the given output has *not* been spent, i.e. is a member of the current UTXO
  40         /// set.
  41         fn is_output_unspent<'a>(&'a self, outpoint: OutPoint) -> AsyncBlockSourceResult<'a, bool>;
  42 }
  43
  44 /// A generic trait which is able to spawn futures in the background.
  45 ///
  46 /// If the `tokio` feature is enabled, this is implemented on `TokioSpawner` struct which
  47 /// delegates to `tokio::spawn()`.
  48 pub trait FutureSpawner : Send + Sync + 'static {
  49         /// Spawns the given future as a background task.
  50         ///
  51         /// This method MUST NOT block on the given future immediately.
  52         fn spawn<T: Future<Output = ()> + Send + 'static>(&self, future: T);
  53 }
  54
  55 #[cfg(feature = "tokio")]
  56 /// A trivial [`FutureSpawner`] which delegates to `tokio::spawn`.
  57 pub struct TokioSpawner;
  58 #[cfg(feature = "tokio")]
  59 impl FutureSpawner for TokioSpawner {
  60         fn spawn<T: Future<Output = ()> + Send + 'static>(&self, future: T) {
  61                 tokio::spawn(future);
  62         }
  63 }
  64
  65 /// A trivial future which joins two other futures and polls them at the same time, returning only
  66 /// once both complete.
  67 pub(crate) struct Joiner<
  68         A: Future<Output=Result<(BlockHash, Option<u32>), BlockSourceError>> + Unpin,
  69         B: Future<Output=Result<BlockHash, BlockSourceError>> + Unpin,
  70 > {
  71         pub a: A,
  72         pub b: B,
  73         a_res: Option<(BlockHash, Option<u32>)>,
  74         b_res: Option<BlockHash>,
  75 }
  76
  77 impl<
  78         A: Future<Output=Result<(BlockHash, Option<u32>), BlockSourceError>> + Unpin,
  79         B: Future<Output=Result<BlockHash, BlockSourceError>> + Unpin,
  80 > Joiner<A, B> {
  81         fn new(a: A, b: B) -> Self { Self { a, b, a_res: None, b_res: None } }
  82 }
  83
  84 impl<
  85         A: Future<Output=Result<(BlockHash, Option<u32>), BlockSourceError>> + Unpin,
  86         B: Future<Output=Result<BlockHash, BlockSourceError>> + Unpin,
  87 > Future for Joiner<A, B> {
  88         type Output = Result<((BlockHash, Option<u32>), BlockHash), BlockSourceError>;
  89         fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<Self::Output> {
  90                 if self.a_res.is_none() {
  91                         match Pin::new(&mut self.a).poll(ctx) {
  92                                 Poll::Ready(res) => {
  93                                         if let Ok(ok) = res {
  94                                                 self.a_res = Some(ok);
  95                                         } else {
  96                                                 return Poll::Ready(Err(res.unwrap_err()));
  97                                         }
  98                                 },
  99                                 Poll::Pending => {},
 100                         }
 101                 }
 102                 if self.b_res.is_none() {
 103                         match Pin::new(&mut self.b).poll(ctx) {
 104                                 Poll::Ready(res) => {
 105                                         if let Ok(ok) = res {
 106                                                 self.b_res = Some(ok);
 107                                         } else {
 108                                                 return Poll::Ready(Err(res.unwrap_err()));
 109                                         }
 110
 111                                 },
 112                                 Poll::Pending => {},
 113                         }
 114                 }
 115                 if let Some(b_res) = self.b_res {
 116                         if let Some(a_res) = self.a_res {
 117                                 return Poll::Ready(Ok((a_res, b_res)))
 118                         }
 119                 }
 120                 Poll::Pending
 121         }
 122 }
 123
 124 /// A struct which wraps a [`UtxoSource`] and a few LDK objects and implements the LDK
 125 /// [`UtxoLookup`] trait.
 126 ///
 127 /// Note that if you're using this against a Bitcoin Core REST or RPC server, you likely wish to
 128 /// increase the `rpcworkqueue` setting in Bitcoin Core as LDK attempts to parallelize requests (a
 129 /// value of 1024 should more than suffice), and ensure you have sufficient file descriptors
 130 /// available on both Bitcoin Core and your LDK application for each request to hold its own
 131 /// connection.
 132 pub struct GossipVerifier<S: FutureSpawner,
 133         Blocks: Deref + Send + Sync + 'static + Clone,
 134         L: Deref + Send + Sync + 'static,
 135         APM: Deref + Send + Sync + 'static + Clone,
 136 > where
 137         Blocks::Target: UtxoSource,
 138         L::Target: Logger,
 139         APM::Target: APeerManager,
 140 {
 141         source: Blocks,
 142         peer_manager: APM,
 143         gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>,
 144         spawn: S,
 145         block_cache: Arc<Mutex<VecDeque<(u32, Block)>>>,
 146 }
 147
 148 const BLOCK_CACHE_SIZE: usize = 5;
 149
 150 impl<S: FutureSpawner,
 151         Blocks: Deref + Send + Sync + Clone,
 152         L: Deref + Send + Sync,
 153         APM: Deref + Send + Sync + Clone,
 154 > GossipVerifier<S, Blocks, L, APM> where
 155         Blocks::Target: UtxoSource,
 156         L::Target: Logger,
 157         APM::Target: APeerManager,
 158 {
 159         /// Constructs a new [`GossipVerifier`].
 160         ///
 161         /// This is expected to be given to a [`P2PGossipSync`] (initially constructed with `None` for
 162         /// the UTXO lookup) via [`P2PGossipSync::add_utxo_lookup`].
 163         pub fn new(source: Blocks, spawn: S, gossiper: Arc<P2PGossipSync<Arc<NetworkGraph<L>>, Self, L>>, peer_manager: APM) -> Self {
 164                 Self {
 165                         source, spawn, gossiper, peer_manager,
 166                         block_cache: Arc::new(Mutex::new(VecDeque::with_capacity(BLOCK_CACHE_SIZE))),
 167                 }
 168         }
 169
 170         async fn retrieve_utxo(
 171                 source: Blocks, block_cache: Arc<Mutex<VecDeque<(u32, Block)>>>, short_channel_id: u64
 172         ) -> Result<TxOut, UtxoLookupError> {
 173                 let block_height = (short_channel_id >> 5 * 8) as u32; // block height is most significant three bytes
 174                 let transaction_index = ((short_channel_id >> 2 * 8) & 0xffffff) as u32;
 175                 let output_index = (short_channel_id & 0xffff) as u16;
 176
 177                 let (outpoint, output);
 178
 179                 'tx_found: loop { // Used as a simple goto
 180                         macro_rules! process_block {
 181                                 ($block: expr) => { {
 182                                         if transaction_index as usize >= $block.txdata.len() {
 183                                                 return Err(UtxoLookupError::UnknownTx);
 184                                         }
 185                                         let transaction = &$block.txdata[transaction_index as usize];
 186                                         if output_index as usize >= transaction.output.len() {
 187                                                 return Err(UtxoLookupError::UnknownTx);
 188                                         }
 189
 190                                         outpoint = OutPoint::new(transaction.txid(), output_index.into());
 191                                         output = transaction.output[output_index as usize].clone();
 192                                 } }
 193                         }
 194                         {
 195                                 let recent_blocks = block_cache.lock().unwrap();
 196                                 for (height, block) in recent_blocks.iter() {
 197                                         if *height == block_height {
 198                                                 process_block!(block);
 199                                                 break 'tx_found;
 200                                         }
 201                                 }
 202                         }
 203
 204                         let ((_, tip_height_opt), block_hash) =
 205                                 Joiner::new(source.get_best_block(), source.get_block_hash_by_height(block_height))
 206                                 .await
 207                                 .map_err(|_| UtxoLookupError::UnknownTx)?;
 208                         if let Some(tip_height) = tip_height_opt {
 209                                 // If the block doesn't yet have five confirmations, error out.
 210                                 //
 211                                 // The BOLT spec requires nodes wait for six confirmations before announcing a
 212                                 // channel, and we give them one block of headroom in case we're delayed seeing a
 213                                 // block.
 214                                 if block_height + 5 > tip_height {
 215                                         return Err(UtxoLookupError::UnknownTx);
 216                                 }
 217                         }
 218                         let block_data = source.get_block(&block_hash).await
 219                                 .map_err(|_| UtxoLookupError::UnknownTx)?;
 220                         let block = match block_data {
 221                                 BlockData::HeaderOnly(_) => return Err(UtxoLookupError::UnknownTx),
 222                                 BlockData::FullBlock(block) => block,
 223                         };
 224                         process_block!(block);
 225                         {
 226                                 let mut recent_blocks = block_cache.lock().unwrap();
 227                                 let mut insert = true;
 228                                 for (height, _) in recent_blocks.iter() {
 229                                         if *height == block_height {
 230                                                 insert = false;
 231                                         }
 232                                 }
 233                                 if insert {
 234                                         if recent_blocks.len() >= BLOCK_CACHE_SIZE {
 235                                                 recent_blocks.pop_front();
 236                                         }
 237                                         recent_blocks.push_back((block_height, block));
 238                                 }
 239                         }
 240                         break 'tx_found;
 241                 };
 242                 let outpoint_unspent =
 243                         source.is_output_unspent(outpoint).await.map_err(|_| UtxoLookupError::UnknownTx)?;
 244                 if outpoint_unspent {
 245                         Ok(output)
 246                 } else {
 247                         Err(UtxoLookupError::UnknownTx)
 248                 }
 249         }
 250 }
 251
 252 impl<S: FutureSpawner,
 253         Blocks: Deref + Send + Sync + Clone,
 254         L: Deref + Send + Sync,
 255         APM: Deref + Send + Sync + Clone,
 256 > Deref for GossipVerifier<S, Blocks, L, APM> where
 257         Blocks::Target: UtxoSource,
 258         L::Target: Logger,
 259         APM::Target: APeerManager,
 260 {
 261         type Target = Self;
 262         fn deref(&self) -> &Self { self }
 263 }
 264
 265
 266 impl<S: FutureSpawner,
 267         Blocks: Deref + Send + Sync + Clone,
 268         L: Deref + Send + Sync,
 269         APM: Deref + Send + Sync + Clone,
 270 > UtxoLookup for GossipVerifier<S, Blocks, L, APM> where
 271         Blocks::Target: UtxoSource,
 272         L::Target: Logger,
 273         APM::Target: APeerManager,
 274 {
 275         fn get_utxo(&self, _chain_hash: &ChainHash, short_channel_id: u64) -> UtxoResult {
 276                 let res = UtxoFuture::new();
 277                 let fut = res.clone();
 278                 let source = self.source.clone();
 279                 let gossiper = Arc::clone(&self.gossiper);
 280                 let block_cache = Arc::clone(&self.block_cache);
 281                 let pm = self.peer_manager.clone();
 282                 self.spawn.spawn(async move {
 283                         let res = Self::retrieve_utxo(source, block_cache, short_channel_id).await;
 284                         fut.resolve(gossiper.network_graph(), &*gossiper, res);
 285                         pm.as_ref().process_events();
 286                 });
 287                 UtxoResult::Async(res)
 288         }
 289 }