#![deny(rustdoc::broken_intra_doc_links)]
#![deny(rustdoc::private_intra_doc_links)]
#![deny(missing_docs)]
#![deny(unsafe_code)]
#![cfg_attr(docsrs, feature(doc_auto_cfg))]
#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
pub mod http;
pub mod init;
pub mod poll;
pub mod gossip;
#[cfg(feature = "rest-client")]
pub mod rest;
#[cfg(feature = "rpc-client")]
pub mod rpc;
#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
mod convert;
#[cfg(test)]
mod test_utils;
#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
mod utils;
use crate::poll::{ChainTip, Poll, ValidatedBlockHeader};
use bitcoin::blockdata::block::{Block, Header};
use bitcoin::hash_types::BlockHash;
use bitcoin::pow::Work;
use lightning::chain;
use lightning::chain::Listen;
use std::future::Future;
use std::ops::Deref;
use std::pin::Pin;
pub trait BlockSource : Sync + Send {
fn get_header<'a>(&'a self, header_hash: &'a BlockHash, height_hint: Option<u32>) -> AsyncBlockSourceResult<'a, BlockHeaderData>;
fn get_block<'a>(&'a self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, BlockData>;
fn get_best_block<'a>(&'a self) -> AsyncBlockSourceResult<(BlockHash, Option<u32>)>;
}
pub type BlockSourceResult<T> = Result<T, BlockSourceError>;
pub type AsyncBlockSourceResult<'a, T> = Pin<Box<dyn Future<Output = BlockSourceResult<T>> + 'a + Send>>;
#[derive(Debug)]
pub struct BlockSourceError {
kind: BlockSourceErrorKind,
error: Box<dyn std::error::Error + Send + Sync>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BlockSourceErrorKind {
Persistent,
Transient,
}
impl BlockSourceError {
pub fn persistent<E>(error: E) -> Self
where E: Into<Box<dyn std::error::Error + Send + Sync>> {
Self {
kind: BlockSourceErrorKind::Persistent,
error: error.into(),
}
}
pub fn transient<E>(error: E) -> Self
where E: Into<Box<dyn std::error::Error + Send + Sync>> {
Self {
kind: BlockSourceErrorKind::Transient,
error: error.into(),
}
}
pub fn kind(&self) -> BlockSourceErrorKind {
self.kind
}
pub fn into_inner(self) -> Box<dyn std::error::Error + Send + Sync> {
self.error
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BlockHeaderData {
pub header: Header,
pub height: u32,
pub chainwork: Work,
}
pub enum BlockData {
FullBlock(Block),
HeaderOnly(Header),
}
pub struct SpvClient<'a, P: Poll, C: Cache, L: Deref>
where L::Target: chain::Listen {
chain_tip: ValidatedBlockHeader,
chain_poller: P,
chain_notifier: ChainNotifier<'a, C, L>,
}
pub trait Cache {
fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader>;
fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader);
fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader>;
}
pub type UnboundedCache = std::collections::HashMap<BlockHash, ValidatedBlockHeader>;
impl Cache for UnboundedCache {
fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
self.get(block_hash)
}
fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
self.insert(block_hash, block_header);
}
fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
self.remove(block_hash)
}
}
impl<'a, P: Poll, C: Cache, L: Deref> SpvClient<'a, P, C, L> where L::Target: chain::Listen {
pub fn new(
chain_tip: ValidatedBlockHeader,
chain_poller: P,
header_cache: &'a mut C,
chain_listener: L,
) -> Self {
let chain_notifier = ChainNotifier { header_cache, chain_listener };
Self { chain_tip, chain_poller, chain_notifier }
}
pub async fn poll_best_tip(&mut self) -> BlockSourceResult<(ChainTip, bool)> {
let chain_tip = self.chain_poller.poll_chain_tip(self.chain_tip).await?;
let blocks_connected = match chain_tip {
ChainTip::Common => false,
ChainTip::Better(chain_tip) => {
debug_assert_ne!(chain_tip.block_hash, self.chain_tip.block_hash);
debug_assert!(chain_tip.chainwork > self.chain_tip.chainwork);
self.update_chain_tip(chain_tip).await
},
ChainTip::Worse(chain_tip) => {
debug_assert_ne!(chain_tip.block_hash, self.chain_tip.block_hash);
debug_assert!(chain_tip.chainwork <= self.chain_tip.chainwork);
false
},
};
Ok((chain_tip, blocks_connected))
}
async fn update_chain_tip(&mut self, best_chain_tip: ValidatedBlockHeader) -> bool {
match self.chain_notifier.synchronize_listener(
best_chain_tip, &self.chain_tip, &mut self.chain_poller).await
{
Ok(_) => {
self.chain_tip = best_chain_tip;
true
},
Err((_, Some(chain_tip))) if chain_tip.block_hash != self.chain_tip.block_hash => {
self.chain_tip = chain_tip;
true
},
Err(_) => false,
}
}
}
pub struct ChainNotifier<'a, C: Cache, L: Deref> where L::Target: chain::Listen {
header_cache: &'a mut C,
chain_listener: L,
}
struct ChainDifference {
common_ancestor: ValidatedBlockHeader,
disconnected_blocks: Vec<ValidatedBlockHeader>,
connected_blocks: Vec<ValidatedBlockHeader>,
}
impl<'a, C: Cache, L: Deref> ChainNotifier<'a, C, L> where L::Target: chain::Listen {
async fn synchronize_listener<P: Poll>(
&mut self,
new_header: ValidatedBlockHeader,
old_header: &ValidatedBlockHeader,
chain_poller: &mut P,
) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
let difference = self.find_difference(new_header, old_header, chain_poller).await
.map_err(|e| (e, None))?;
self.disconnect_blocks(difference.disconnected_blocks);
self.connect_blocks(
difference.common_ancestor,
difference.connected_blocks,
chain_poller,
).await
}
async fn find_difference<P: Poll>(
&self,
current_header: ValidatedBlockHeader,
prev_header: &ValidatedBlockHeader,
chain_poller: &mut P,
) -> BlockSourceResult<ChainDifference> {
let mut disconnected_blocks = Vec::new();
let mut connected_blocks = Vec::new();
let mut current = current_header;
let mut previous = *prev_header;
loop {
if current.block_hash == previous.block_hash {
break;
}
let current_height = current.height;
let previous_height = previous.height;
if current_height <= previous_height {
disconnected_blocks.push(previous);
previous = self.look_up_previous_header(chain_poller, &previous).await?;
}
if current_height >= previous_height {
connected_blocks.push(current);
current = self.look_up_previous_header(chain_poller, ¤t).await?;
}
}
let common_ancestor = current;
Ok(ChainDifference { common_ancestor, disconnected_blocks, connected_blocks })
}
async fn look_up_previous_header<P: Poll>(
&self,
chain_poller: &mut P,
header: &ValidatedBlockHeader,
) -> BlockSourceResult<ValidatedBlockHeader> {
match self.header_cache.look_up(&header.header.prev_blockhash) {
Some(prev_header) => Ok(*prev_header),
None => chain_poller.look_up_previous_header(header).await,
}
}
fn disconnect_blocks(&mut self, mut disconnected_blocks: Vec<ValidatedBlockHeader>) {
for header in disconnected_blocks.drain(..) {
if let Some(cached_header) = self.header_cache.block_disconnected(&header.block_hash) {
assert_eq!(cached_header, header);
}
self.chain_listener.block_disconnected(&header.header, header.height);
}
}
async fn connect_blocks<P: Poll>(
&mut self,
mut new_tip: ValidatedBlockHeader,
mut connected_blocks: Vec<ValidatedBlockHeader>,
chain_poller: &mut P,
) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
for header in connected_blocks.drain(..).rev() {
let height = header.height;
let block_data = chain_poller
.fetch_block(&header).await
.map_err(|e| (e, Some(new_tip)))?;
debug_assert_eq!(block_data.block_hash, header.block_hash);
match block_data.deref() {
BlockData::FullBlock(block) => {
self.chain_listener.block_connected(block, height);
},
BlockData::HeaderOnly(header) => {
self.chain_listener.filtered_block_connected(header, &[], height);
},
}
self.header_cache.block_connected(header.block_hash, header);
new_tip = header;
}
Ok(())
}
}
#[cfg(test)]
mod spv_client_tests {
use crate::test_utils::{Blockchain, NullChainListener};
use super::*;
use bitcoin::network::constants::Network;
#[tokio::test]
async fn poll_from_chain_without_headers() {
let mut chain = Blockchain::default().with_height(3).without_headers();
let best_tip = chain.at_height(1);
let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
let mut cache = UnboundedCache::new();
let mut listener = NullChainListener {};
let mut client = SpvClient::new(best_tip, poller, &mut cache, &mut listener);
match client.poll_best_tip().await {
Err(e) => {
assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
assert_eq!(e.into_inner().as_ref().to_string(), "header not found");
},
Ok(_) => panic!("Expected error"),
}
assert_eq!(client.chain_tip, best_tip);
}
#[tokio::test]
async fn poll_from_chain_with_common_tip() {
let mut chain = Blockchain::default().with_height(3);
let common_tip = chain.tip();
let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
let mut cache = UnboundedCache::new();
let mut listener = NullChainListener {};
let mut client = SpvClient::new(common_tip, poller, &mut cache, &mut listener);
match client.poll_best_tip().await {
Err(e) => panic!("Unexpected error: {:?}", e),
Ok((chain_tip, blocks_connected)) => {
assert_eq!(chain_tip, ChainTip::Common);
assert!(!blocks_connected);
},
}
assert_eq!(client.chain_tip, common_tip);
}
#[tokio::test]
async fn poll_from_chain_with_better_tip() {
let mut chain = Blockchain::default().with_height(3);
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
let mut cache = UnboundedCache::new();
let mut listener = NullChainListener {};
let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
match client.poll_best_tip().await {
Err(e) => panic!("Unexpected error: {:?}", e),
Ok((chain_tip, blocks_connected)) => {
assert_eq!(chain_tip, ChainTip::Better(new_tip));
assert!(blocks_connected);
},
}
assert_eq!(client.chain_tip, new_tip);
}
#[tokio::test]
async fn poll_from_chain_with_better_tip_and_without_any_new_blocks() {
let mut chain = Blockchain::default().with_height(3).without_blocks(2..);
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
let mut cache = UnboundedCache::new();
let mut listener = NullChainListener {};
let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
match client.poll_best_tip().await {
Err(e) => panic!("Unexpected error: {:?}", e),
Ok((chain_tip, blocks_connected)) => {
assert_eq!(chain_tip, ChainTip::Better(new_tip));
assert!(!blocks_connected);
},
}
assert_eq!(client.chain_tip, old_tip);
}
#[tokio::test]
async fn poll_from_chain_with_better_tip_and_without_some_new_blocks() {
let mut chain = Blockchain::default().with_height(3).without_blocks(3..);
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
let mut cache = UnboundedCache::new();
let mut listener = NullChainListener {};
let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
match client.poll_best_tip().await {
Err(e) => panic!("Unexpected error: {:?}", e),
Ok((chain_tip, blocks_connected)) => {
assert_eq!(chain_tip, ChainTip::Better(new_tip));
assert!(blocks_connected);
},
}
assert_eq!(client.chain_tip, chain.at_height(2));
}
#[tokio::test]
async fn poll_from_chain_with_worse_tip() {
let mut chain = Blockchain::default().with_height(3);
let best_tip = chain.tip();
chain.disconnect_tip();
let worse_tip = chain.tip();
let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
let mut cache = UnboundedCache::new();
let mut listener = NullChainListener {};
let mut client = SpvClient::new(best_tip, poller, &mut cache, &mut listener);
match client.poll_best_tip().await {
Err(e) => panic!("Unexpected error: {:?}", e),
Ok((chain_tip, blocks_connected)) => {
assert_eq!(chain_tip, ChainTip::Worse(worse_tip));
assert!(!blocks_connected);
},
}
assert_eq!(client.chain_tip, best_tip);
}
}
#[cfg(test)]
mod chain_notifier_tests {
use crate::test_utils::{Blockchain, MockChainListener};
use super::*;
use bitcoin::network::constants::Network;
#[tokio::test]
async fn sync_from_same_chain() {
let mut chain = Blockchain::default().with_height(3);
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let chain_listener = &MockChainListener::new()
.expect_block_connected(*chain.at_height(2))
.expect_block_connected(*new_tip);
let mut notifier = ChainNotifier {
header_cache: &mut chain.header_cache(0..=1),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((e, _)) => panic!("Unexpected error: {:?}", e),
Ok(_) => {},
}
}
#[tokio::test]
async fn sync_from_different_chains() {
let mut test_chain = Blockchain::with_network(Network::Testnet).with_height(1);
let main_chain = Blockchain::with_network(Network::Bitcoin).with_height(1);
let new_tip = test_chain.tip();
let old_tip = main_chain.tip();
let chain_listener = &MockChainListener::new();
let mut notifier = ChainNotifier {
header_cache: &mut main_chain.header_cache(0..=1),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut test_chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((e, _)) => {
assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
assert_eq!(e.into_inner().as_ref().to_string(), "genesis block reached");
},
Ok(_) => panic!("Expected error"),
}
}
#[tokio::test]
async fn sync_from_equal_length_fork() {
let main_chain = Blockchain::default().with_height(2);
let mut fork_chain = main_chain.fork_at_height(1);
let new_tip = fork_chain.tip();
let old_tip = main_chain.tip();
let chain_listener = &MockChainListener::new()
.expect_block_disconnected(*old_tip)
.expect_block_connected(*new_tip);
let mut notifier = ChainNotifier {
header_cache: &mut main_chain.header_cache(0..=2),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((e, _)) => panic!("Unexpected error: {:?}", e),
Ok(_) => {},
}
}
#[tokio::test]
async fn sync_from_shorter_fork() {
let main_chain = Blockchain::default().with_height(3);
let mut fork_chain = main_chain.fork_at_height(1);
fork_chain.disconnect_tip();
let new_tip = fork_chain.tip();
let old_tip = main_chain.tip();
let chain_listener = &MockChainListener::new()
.expect_block_disconnected(*old_tip)
.expect_block_disconnected(*main_chain.at_height(2))
.expect_block_connected(*new_tip);
let mut notifier = ChainNotifier {
header_cache: &mut main_chain.header_cache(0..=3),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((e, _)) => panic!("Unexpected error: {:?}", e),
Ok(_) => {},
}
}
#[tokio::test]
async fn sync_from_longer_fork() {
let mut main_chain = Blockchain::default().with_height(3);
let mut fork_chain = main_chain.fork_at_height(1);
main_chain.disconnect_tip();
let new_tip = fork_chain.tip();
let old_tip = main_chain.tip();
let chain_listener = &MockChainListener::new()
.expect_block_disconnected(*old_tip)
.expect_block_connected(*fork_chain.at_height(2))
.expect_block_connected(*new_tip);
let mut notifier = ChainNotifier {
header_cache: &mut main_chain.header_cache(0..=2),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((e, _)) => panic!("Unexpected error: {:?}", e),
Ok(_) => {},
}
}
#[tokio::test]
async fn sync_from_chain_without_headers() {
let mut chain = Blockchain::default().with_height(3).without_headers();
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let chain_listener = &MockChainListener::new();
let mut notifier = ChainNotifier {
header_cache: &mut chain.header_cache(0..=1),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((_, tip)) => assert_eq!(tip, None),
Ok(_) => panic!("Expected error"),
}
}
#[tokio::test]
async fn sync_from_chain_without_any_new_blocks() {
let mut chain = Blockchain::default().with_height(3).without_blocks(2..);
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let chain_listener = &MockChainListener::new();
let mut notifier = ChainNotifier {
header_cache: &mut chain.header_cache(0..=3),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((_, tip)) => assert_eq!(tip, Some(old_tip)),
Ok(_) => panic!("Expected error"),
}
}
#[tokio::test]
async fn sync_from_chain_without_some_new_blocks() {
let mut chain = Blockchain::default().with_height(3).without_blocks(3..);
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let chain_listener = &MockChainListener::new()
.expect_block_connected(*chain.at_height(2));
let mut notifier = ChainNotifier {
header_cache: &mut chain.header_cache(0..=3),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((_, tip)) => assert_eq!(tip, Some(chain.at_height(2))),
Ok(_) => panic!("Expected error"),
}
}
#[tokio::test]
async fn sync_from_chain_with_filtered_blocks() {
let mut chain = Blockchain::default().with_height(3).filtered_blocks();
let new_tip = chain.tip();
let old_tip = chain.at_height(1);
let chain_listener = &MockChainListener::new()
.expect_filtered_block_connected(*chain.at_height(2))
.expect_filtered_block_connected(*new_tip);
let mut notifier = ChainNotifier {
header_cache: &mut chain.header_cache(0..=1),
chain_listener,
};
let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
Err((e, _)) => panic!("Unexpected error: {:?}", e),
Ok(_) => {},
}
}
}