1#![deny(rustdoc::broken_intra_doc_links)]
17#![deny(rustdoc::private_intra_doc_links)]
18#![deny(missing_docs)]
19#![deny(unsafe_code)]
20#![cfg_attr(docsrs, feature(doc_cfg))]
21
22#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
23pub mod http;
24
25pub mod init;
26pub mod poll;
27
28pub mod gossip;
29
30#[cfg(feature = "rest-client")]
31pub mod rest;
32
33#[cfg(feature = "rpc-client")]
34pub mod rpc;
35
36#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
37mod convert;
38
39#[cfg(test)]
40mod test_utils;
41
42#[cfg(any(feature = "rest-client", feature = "rpc-client"))]
43mod utils;
44
45use crate::poll::{ChainTip, Poll, ValidatedBlockHeader};
46
47use bitcoin::block::{Block, Header};
48use bitcoin::hash_types::BlockHash;
49use bitcoin::pow::Work;
50
51use lightning::chain;
52use lightning::chain::{BestBlock, Listen};
53
54use std::future::Future;
55use std::ops::Deref;
56use std::pin::Pin;
57
58pub trait BlockSource: Sync + Send {
60 fn get_header<'a>(
67 &'a self, header_hash: &'a BlockHash, height_hint: Option<u32>,
68 ) -> AsyncBlockSourceResult<'a, BlockHeaderData>;
69
70 fn get_block<'a>(&'a self, header_hash: &'a BlockHash)
73 -> AsyncBlockSourceResult<'a, BlockData>;
74
75 fn get_best_block(&self) -> AsyncBlockSourceResult<'_, (BlockHash, Option<u32>)>;
82}
83
84pub type BlockSourceResult<T> = Result<T, BlockSourceError>;
86
87pub type AsyncBlockSourceResult<'a, T> =
91 Pin<Box<dyn Future<Output = BlockSourceResult<T>> + 'a + Send>>;
92
93#[derive(Debug)]
98pub struct BlockSourceError {
99 kind: BlockSourceErrorKind,
100 error: Box<dyn std::error::Error + Send + Sync>,
101}
102
103#[derive(Clone, Copy, Debug, PartialEq, Eq)]
105pub enum BlockSourceErrorKind {
106 Persistent,
108
109 Transient,
111}
112
113impl BlockSourceError {
114 pub fn persistent<E>(error: E) -> Self
116 where
117 E: Into<Box<dyn std::error::Error + Send + Sync>>,
118 {
119 Self { kind: BlockSourceErrorKind::Persistent, error: error.into() }
120 }
121
122 pub fn transient<E>(error: E) -> Self
124 where
125 E: Into<Box<dyn std::error::Error + Send + Sync>>,
126 {
127 Self { kind: BlockSourceErrorKind::Transient, error: error.into() }
128 }
129
130 pub fn kind(&self) -> BlockSourceErrorKind {
132 self.kind
133 }
134
135 pub fn into_inner(self) -> Box<dyn std::error::Error + Send + Sync> {
140 self.error
141 }
142}
143
144#[derive(Clone, Copy, Debug, PartialEq, Eq)]
147pub struct BlockHeaderData {
148 pub header: Header,
150
151 pub height: u32,
153
154 pub chainwork: Work,
156}
157
158pub enum BlockData {
164 FullBlock(Block),
166 HeaderOnly(Header),
168}
169
170pub struct SpvClient<'a, P: Poll, C: Cache, L: Deref>
183where
184 L::Target: chain::Listen,
185{
186 chain_tip: ValidatedBlockHeader,
187 chain_poller: P,
188 chain_notifier: ChainNotifier<'a, C, L>,
189}
190
191pub trait Cache {
202 fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader>;
204
205 fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader);
208
209 fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader>;
212}
213
214pub type UnboundedCache = std::collections::HashMap<BlockHash, ValidatedBlockHeader>;
216
217impl Cache for UnboundedCache {
218 fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
219 self.get(block_hash)
220 }
221
222 fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
223 self.insert(block_hash, block_header);
224 }
225
226 fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
227 self.remove(block_hash)
228 }
229}
230
231impl<'a, P: Poll, C: Cache, L: Deref> SpvClient<'a, P, C, L>
232where
233 L::Target: chain::Listen,
234{
235 pub fn new(
246 chain_tip: ValidatedBlockHeader, chain_poller: P, header_cache: &'a mut C,
247 chain_listener: L,
248 ) -> Self {
249 let chain_notifier = ChainNotifier { header_cache, chain_listener };
250 Self { chain_tip, chain_poller, chain_notifier }
251 }
252
253 pub async fn poll_best_tip(&mut self) -> BlockSourceResult<(ChainTip, bool)> {
259 let chain_tip = self.chain_poller.poll_chain_tip(self.chain_tip).await?;
260 let blocks_connected = match chain_tip {
261 ChainTip::Common => false,
262 ChainTip::Better(chain_tip) => {
263 debug_assert_ne!(chain_tip.block_hash, self.chain_tip.block_hash);
264 debug_assert!(chain_tip.chainwork > self.chain_tip.chainwork);
265 self.update_chain_tip(chain_tip).await
266 },
267 ChainTip::Worse(chain_tip) => {
268 debug_assert_ne!(chain_tip.block_hash, self.chain_tip.block_hash);
269 debug_assert!(chain_tip.chainwork <= self.chain_tip.chainwork);
270 false
271 },
272 };
273 Ok((chain_tip, blocks_connected))
274 }
275
276 async fn update_chain_tip(&mut self, best_chain_tip: ValidatedBlockHeader) -> bool {
279 match self
280 .chain_notifier
281 .synchronize_listener(best_chain_tip, &self.chain_tip, &mut self.chain_poller)
282 .await
283 {
284 Ok(_) => {
285 self.chain_tip = best_chain_tip;
286 true
287 },
288 Err((_, Some(chain_tip))) if chain_tip.block_hash != self.chain_tip.block_hash => {
289 self.chain_tip = chain_tip;
290 true
291 },
292 Err(_) => false,
293 }
294 }
295}
296
297pub struct ChainNotifier<'a, C: Cache, L: Deref>
301where
302 L::Target: chain::Listen,
303{
304 header_cache: &'a mut C,
306
307 chain_listener: L,
309}
310
311struct ChainDifference {
317 common_ancestor: ValidatedBlockHeader,
321
322 disconnected_blocks: Vec<ValidatedBlockHeader>,
324
325 connected_blocks: Vec<ValidatedBlockHeader>,
327}
328
329impl<'a, C: Cache, L: Deref> ChainNotifier<'a, C, L>
330where
331 L::Target: chain::Listen,
332{
333 async fn synchronize_listener<P: Poll>(
341 &mut self, new_header: ValidatedBlockHeader, old_header: &ValidatedBlockHeader,
342 chain_poller: &mut P,
343 ) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
344 let difference = self
345 .find_difference(new_header, old_header, chain_poller)
346 .await
347 .map_err(|e| (e, None))?;
348 self.disconnect_blocks(difference.disconnected_blocks);
349 self.connect_blocks(difference.common_ancestor, difference.connected_blocks, chain_poller)
350 .await
351 }
352
353 async fn find_difference<P: Poll>(
358 &self, current_header: ValidatedBlockHeader, prev_header: &ValidatedBlockHeader,
359 chain_poller: &mut P,
360 ) -> BlockSourceResult<ChainDifference> {
361 let mut disconnected_blocks = Vec::new();
362 let mut connected_blocks = Vec::new();
363 let mut current = current_header;
364 let mut previous = *prev_header;
365 loop {
366 if current.block_hash == previous.block_hash {
368 break;
369 }
370
371 let current_height = current.height;
374 let previous_height = previous.height;
375 if current_height <= previous_height {
376 disconnected_blocks.push(previous);
377 previous = self.look_up_previous_header(chain_poller, &previous).await?;
378 }
379 if current_height >= previous_height {
380 connected_blocks.push(current);
381 current = self.look_up_previous_header(chain_poller, ¤t).await?;
382 }
383 }
384
385 let common_ancestor = current;
386 Ok(ChainDifference { common_ancestor, disconnected_blocks, connected_blocks })
387 }
388
389 async fn look_up_previous_header<P: Poll>(
392 &self, chain_poller: &mut P, header: &ValidatedBlockHeader,
393 ) -> BlockSourceResult<ValidatedBlockHeader> {
394 match self.header_cache.look_up(&header.header.prev_blockhash) {
395 Some(prev_header) => Ok(*prev_header),
396 None => chain_poller.look_up_previous_header(header).await,
397 }
398 }
399
400 fn disconnect_blocks(&mut self, disconnected_blocks: Vec<ValidatedBlockHeader>) {
402 for header in disconnected_blocks.iter() {
403 if let Some(cached_header) = self.header_cache.block_disconnected(&header.block_hash) {
404 assert_eq!(cached_header, *header);
405 }
406 }
407 if let Some(block) = disconnected_blocks.last() {
408 let fork_point = BestBlock::new(block.header.prev_blockhash, block.height - 1);
409 self.chain_listener.blocks_disconnected(fork_point);
410 }
411 }
412
413 async fn connect_blocks<P: Poll>(
415 &mut self, mut new_tip: ValidatedBlockHeader,
416 mut connected_blocks: Vec<ValidatedBlockHeader>, chain_poller: &mut P,
417 ) -> Result<(), (BlockSourceError, Option<ValidatedBlockHeader>)> {
418 for header in connected_blocks.drain(..).rev() {
419 let height = header.height;
420 let block_data =
421 chain_poller.fetch_block(&header).await.map_err(|e| (e, Some(new_tip)))?;
422 debug_assert_eq!(block_data.block_hash, header.block_hash);
423
424 match block_data.deref() {
425 BlockData::FullBlock(block) => {
426 self.chain_listener.block_connected(block, height);
427 },
428 BlockData::HeaderOnly(header) => {
429 self.chain_listener.filtered_block_connected(header, &[], height);
430 },
431 }
432
433 self.header_cache.block_connected(header.block_hash, header);
434 new_tip = header;
435 }
436
437 Ok(())
438 }
439}
440
441#[cfg(test)]
442mod spv_client_tests {
443 use super::*;
444 use crate::test_utils::{Blockchain, NullChainListener};
445
446 use bitcoin::network::Network;
447
448 #[tokio::test]
449 async fn poll_from_chain_without_headers() {
450 let mut chain = Blockchain::default().with_height(3).without_headers();
451 let best_tip = chain.at_height(1);
452
453 let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
454 let mut cache = UnboundedCache::new();
455 let mut listener = NullChainListener {};
456 let mut client = SpvClient::new(best_tip, poller, &mut cache, &mut listener);
457 match client.poll_best_tip().await {
458 Err(e) => {
459 assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
460 assert_eq!(e.into_inner().as_ref().to_string(), "header not found");
461 },
462 Ok(_) => panic!("Expected error"),
463 }
464 assert_eq!(client.chain_tip, best_tip);
465 }
466
467 #[tokio::test]
468 async fn poll_from_chain_with_common_tip() {
469 let mut chain = Blockchain::default().with_height(3);
470 let common_tip = chain.tip();
471
472 let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
473 let mut cache = UnboundedCache::new();
474 let mut listener = NullChainListener {};
475 let mut client = SpvClient::new(common_tip, poller, &mut cache, &mut listener);
476 match client.poll_best_tip().await {
477 Err(e) => panic!("Unexpected error: {:?}", e),
478 Ok((chain_tip, blocks_connected)) => {
479 assert_eq!(chain_tip, ChainTip::Common);
480 assert!(!blocks_connected);
481 },
482 }
483 assert_eq!(client.chain_tip, common_tip);
484 }
485
486 #[tokio::test]
487 async fn poll_from_chain_with_better_tip() {
488 let mut chain = Blockchain::default().with_height(3);
489 let new_tip = chain.tip();
490 let old_tip = chain.at_height(1);
491
492 let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
493 let mut cache = UnboundedCache::new();
494 let mut listener = NullChainListener {};
495 let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
496 match client.poll_best_tip().await {
497 Err(e) => panic!("Unexpected error: {:?}", e),
498 Ok((chain_tip, blocks_connected)) => {
499 assert_eq!(chain_tip, ChainTip::Better(new_tip));
500 assert!(blocks_connected);
501 },
502 }
503 assert_eq!(client.chain_tip, new_tip);
504 }
505
506 #[tokio::test]
507 async fn poll_from_chain_with_better_tip_and_without_any_new_blocks() {
508 let mut chain = Blockchain::default().with_height(3).without_blocks(2..);
509 let new_tip = chain.tip();
510 let old_tip = chain.at_height(1);
511
512 let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
513 let mut cache = UnboundedCache::new();
514 let mut listener = NullChainListener {};
515 let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
516 match client.poll_best_tip().await {
517 Err(e) => panic!("Unexpected error: {:?}", e),
518 Ok((chain_tip, blocks_connected)) => {
519 assert_eq!(chain_tip, ChainTip::Better(new_tip));
520 assert!(!blocks_connected);
521 },
522 }
523 assert_eq!(client.chain_tip, old_tip);
524 }
525
526 #[tokio::test]
527 async fn poll_from_chain_with_better_tip_and_without_some_new_blocks() {
528 let mut chain = Blockchain::default().with_height(3).without_blocks(3..);
529 let new_tip = chain.tip();
530 let old_tip = chain.at_height(1);
531
532 let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
533 let mut cache = UnboundedCache::new();
534 let mut listener = NullChainListener {};
535 let mut client = SpvClient::new(old_tip, poller, &mut cache, &mut listener);
536 match client.poll_best_tip().await {
537 Err(e) => panic!("Unexpected error: {:?}", e),
538 Ok((chain_tip, blocks_connected)) => {
539 assert_eq!(chain_tip, ChainTip::Better(new_tip));
540 assert!(blocks_connected);
541 },
542 }
543 assert_eq!(client.chain_tip, chain.at_height(2));
544 }
545
546 #[tokio::test]
547 async fn poll_from_chain_with_worse_tip() {
548 let mut chain = Blockchain::default().with_height(3);
549 let best_tip = chain.tip();
550 chain.disconnect_tip();
551 let worse_tip = chain.tip();
552
553 let poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
554 let mut cache = UnboundedCache::new();
555 let mut listener = NullChainListener {};
556 let mut client = SpvClient::new(best_tip, poller, &mut cache, &mut listener);
557 match client.poll_best_tip().await {
558 Err(e) => panic!("Unexpected error: {:?}", e),
559 Ok((chain_tip, blocks_connected)) => {
560 assert_eq!(chain_tip, ChainTip::Worse(worse_tip));
561 assert!(!blocks_connected);
562 },
563 }
564 assert_eq!(client.chain_tip, best_tip);
565 }
566}
567
568#[cfg(test)]
569mod chain_notifier_tests {
570 use super::*;
571 use crate::test_utils::{Blockchain, MockChainListener};
572
573 use bitcoin::network::Network;
574
575 #[tokio::test]
576 async fn sync_from_same_chain() {
577 let mut chain = Blockchain::default().with_height(3);
578
579 let new_tip = chain.tip();
580 let old_tip = chain.at_height(1);
581 let chain_listener = &MockChainListener::new()
582 .expect_block_connected(*chain.at_height(2))
583 .expect_block_connected(*new_tip);
584 let mut notifier =
585 ChainNotifier { header_cache: &mut chain.header_cache(0..=1), chain_listener };
586 let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
587 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
588 Err((e, _)) => panic!("Unexpected error: {:?}", e),
589 Ok(_) => {},
590 }
591 }
592
593 #[tokio::test]
594 async fn sync_from_different_chains() {
595 let mut test_chain = Blockchain::with_network(Network::Testnet).with_height(1);
596 let main_chain = Blockchain::with_network(Network::Bitcoin).with_height(1);
597
598 let new_tip = test_chain.tip();
599 let old_tip = main_chain.tip();
600 let chain_listener = &MockChainListener::new();
601 let mut notifier =
602 ChainNotifier { header_cache: &mut main_chain.header_cache(0..=1), chain_listener };
603 let mut poller = poll::ChainPoller::new(&mut test_chain, Network::Testnet);
604 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
605 Err((e, _)) => {
606 assert_eq!(e.kind(), BlockSourceErrorKind::Persistent);
607 assert_eq!(e.into_inner().as_ref().to_string(), "genesis block reached");
608 },
609 Ok(_) => panic!("Expected error"),
610 }
611 }
612
613 #[tokio::test]
614 async fn sync_from_equal_length_fork() {
615 let main_chain = Blockchain::default().with_height(2);
616 let mut fork_chain = main_chain.fork_at_height(1);
617
618 let new_tip = fork_chain.tip();
619 let old_tip = main_chain.tip();
620 let chain_listener = &MockChainListener::new()
621 .expect_blocks_disconnected(*fork_chain.at_height(1))
622 .expect_block_connected(*new_tip);
623 let mut notifier =
624 ChainNotifier { header_cache: &mut main_chain.header_cache(0..=2), chain_listener };
625 let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
626 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
627 Err((e, _)) => panic!("Unexpected error: {:?}", e),
628 Ok(_) => {},
629 }
630 }
631
632 #[tokio::test]
633 async fn sync_from_shorter_fork() {
634 let main_chain = Blockchain::default().with_height(3);
635 let mut fork_chain = main_chain.fork_at_height(1);
636 fork_chain.disconnect_tip();
637
638 let new_tip = fork_chain.tip();
639 let old_tip = main_chain.tip();
640 let chain_listener = &MockChainListener::new()
641 .expect_blocks_disconnected(*main_chain.at_height(1))
642 .expect_block_connected(*new_tip);
643 let mut notifier =
644 ChainNotifier { header_cache: &mut main_chain.header_cache(0..=3), chain_listener };
645 let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
646 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
647 Err((e, _)) => panic!("Unexpected error: {:?}", e),
648 Ok(_) => {},
649 }
650 }
651
652 #[tokio::test]
653 async fn sync_from_longer_fork() {
654 let mut main_chain = Blockchain::default().with_height(3);
655 let mut fork_chain = main_chain.fork_at_height(1);
656 main_chain.disconnect_tip();
657
658 let new_tip = fork_chain.tip();
659 let old_tip = main_chain.tip();
660 let chain_listener = &MockChainListener::new()
661 .expect_blocks_disconnected(*fork_chain.at_height(1))
662 .expect_block_connected(*fork_chain.at_height(2))
663 .expect_block_connected(*new_tip);
664 let mut notifier =
665 ChainNotifier { header_cache: &mut main_chain.header_cache(0..=2), chain_listener };
666 let mut poller = poll::ChainPoller::new(&mut fork_chain, Network::Testnet);
667 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
668 Err((e, _)) => panic!("Unexpected error: {:?}", e),
669 Ok(_) => {},
670 }
671 }
672
673 #[tokio::test]
674 async fn sync_from_chain_without_headers() {
675 let mut chain = Blockchain::default().with_height(3).without_headers();
676
677 let new_tip = chain.tip();
678 let old_tip = chain.at_height(1);
679 let chain_listener = &MockChainListener::new();
680 let mut notifier =
681 ChainNotifier { header_cache: &mut chain.header_cache(0..=1), chain_listener };
682 let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
683 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
684 Err((_, tip)) => assert_eq!(tip, None),
685 Ok(_) => panic!("Expected error"),
686 }
687 }
688
689 #[tokio::test]
690 async fn sync_from_chain_without_any_new_blocks() {
691 let mut chain = Blockchain::default().with_height(3).without_blocks(2..);
692
693 let new_tip = chain.tip();
694 let old_tip = chain.at_height(1);
695 let chain_listener = &MockChainListener::new();
696 let mut notifier =
697 ChainNotifier { header_cache: &mut chain.header_cache(0..=3), chain_listener };
698 let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
699 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
700 Err((_, tip)) => assert_eq!(tip, Some(old_tip)),
701 Ok(_) => panic!("Expected error"),
702 }
703 }
704
705 #[tokio::test]
706 async fn sync_from_chain_without_some_new_blocks() {
707 let mut chain = Blockchain::default().with_height(3).without_blocks(3..);
708
709 let new_tip = chain.tip();
710 let old_tip = chain.at_height(1);
711 let chain_listener = &MockChainListener::new().expect_block_connected(*chain.at_height(2));
712 let mut notifier =
713 ChainNotifier { header_cache: &mut chain.header_cache(0..=3), chain_listener };
714 let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
715 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
716 Err((_, tip)) => assert_eq!(tip, Some(chain.at_height(2))),
717 Ok(_) => panic!("Expected error"),
718 }
719 }
720
721 #[tokio::test]
722 async fn sync_from_chain_with_filtered_blocks() {
723 let mut chain = Blockchain::default().with_height(3).filtered_blocks();
724
725 let new_tip = chain.tip();
726 let old_tip = chain.at_height(1);
727 let chain_listener = &MockChainListener::new()
728 .expect_filtered_block_connected(*chain.at_height(2))
729 .expect_filtered_block_connected(*new_tip);
730 let mut notifier =
731 ChainNotifier { header_cache: &mut chain.header_cache(0..=1), chain_listener };
732 let mut poller = poll::ChainPoller::new(&mut chain, Network::Testnet);
733 match notifier.synchronize_listener(new_tip, &old_tip, &mut poller).await {
734 Err((e, _)) => panic!("Unexpected error: {:?}", e),
735 Ok(_) => {},
736 }
737 }
738}