From ec02e0387a8c084a634640eb3b27e2b4aa84034e Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Fri, 2 Dec 2022 22:54:41 +0100 Subject: [PATCH 1/8] Various small fixes The 'sync' event wasn't firing on every block, now it should be. --- client/src/event.rs | 4 +++- net/poll/src/reactor.rs | 4 +--- p2p/src/fsm/cbfmgr.rs | 11 +++++++---- 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/client/src/event.rs b/client/src/event.rs index 1b084ab1..56adef8a 100644 --- a/client/src/event.rs +++ b/client/src/event.rs @@ -469,13 +469,15 @@ impl Mapper { // If we have no blocks left to process, we are synced to the height of the last // processed filter. Otherwise, we're synced up to the last processed block. let height = if self.pending.is_empty() { - self.filter_height + self.filter_height.max(self.block_height) } else { self.block_height }; // Ensure we only broadcast sync events when the sync height has changed. if height > self.sync_height { + debug_assert!(height == self.sync_height + 1); + self.sync_height = height; emitter.emit(Event::Synced { diff --git a/net/poll/src/reactor.rs b/net/poll/src/reactor.rs index e8eb547f..9387617f 100644 --- a/net/poll/src/reactor.rs +++ b/net/poll/src/reactor.rs @@ -305,7 +305,7 @@ impl Reactor { match self::dial(&socket_addr) { Ok(stream) => { - trace!("{:#?}", stream); + trace!("Stream established with {}", socket_addr); self.register_peer(addr.clone(), stream, Link::Outbound); self.connecting.insert(addr.clone()); @@ -340,8 +340,6 @@ impl Reactor { self.timeouts.register((), local_time + timeout); } Io::Event(event) => { - trace!("Event: {:?}", event); - publisher.publish(event); } } diff --git a/p2p/src/fsm/cbfmgr.rs b/p2p/src/fsm/cbfmgr.rs index e563e7b6..69ebc085 100644 --- a/p2p/src/fsm/cbfmgr.rs +++ b/p2p/src/fsm/cbfmgr.rs @@ -173,7 +173,7 @@ impl std::fmt::Display for Event { } => { write!( fmt, - "Filter processed at height {} (match = {}, valid = {})", + "Filter {} processed (match = {}, valid = {})", height, matched, valid ) } @@ -415,7 +415,8 @@ impl + SetTimer + Disconnect, C: Clock> FilterManager } log::debug!( - "[spv] Rollback from {} to {}, start = {}, height = {}", + target: "spv", + "Rollback from {} to {}, start = {}, height = {}", current, self.rescan.current, start, @@ -536,7 +537,8 @@ impl + SetTimer + Disconnect, C: Clock> FilterManager let timeout = self.config.request_timeout; log::debug!( - "Requested filter(s) in range {} to {} from {} (stop = {})", + target: "spv", + "Requesting filter(s) in range {} to {} from {} (stop = {})", range.start(), range.end(), peer, @@ -563,7 +565,8 @@ impl + SetTimer + Disconnect, C: Clock> FilterManager let stop_hash = msg.stop_hash; log::debug!( - "[spv] Received {} filter header(s) from {}", + target: "spv", + "Received {} filter header(s) from {}", msg.filter_hashes.len(), from ); From cd8f34c61a6bceaf25d67eca4b399c02d85dee92 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sat, 3 Dec 2022 00:39:49 +0100 Subject: [PATCH 2/8] [client] Pass `Config` in `Client::new` This lets us include the block hash in the `Synced` event. --- client/src/client.rs | 17 ++++++++--------- client/src/event.rs | 45 ++++++++++++++++++++++---------------------- node/src/lib.rs | 2 +- src/lib.rs | 4 ++-- wallet/src/lib.rs | 4 ++-- wallet/src/wallet.rs | 2 +- 6 files changed, 37 insertions(+), 37 deletions(-) diff --git a/client/src/client.rs b/client/src/client.rs index 3b7a7726..68fc8074 100644 --- a/client/src/client.rs +++ b/client/src/client.rs @@ -189,6 +189,7 @@ impl ClientRunner { /// A light-client process. pub struct Client { handle: Handle, + config: Config, commands: chan::Receiver, publisher: Publisher, reactor: R, @@ -196,7 +197,7 @@ pub struct Client { impl Client { /// Create a new client. - pub fn new() -> Result { + pub fn new(config: Config) -> Result { let (commands_tx, commands_rx) = chan::unbounded::(); let (event_pub, events) = event::broadcast(|e, p| p.emit(e)); let (blocks_pub, blocks) = event::broadcast(|e, p| { @@ -221,7 +222,7 @@ impl Client { } }); let (publisher, subscriber) = event::broadcast({ - let mut mapper = Mapper::default(); + let mut mapper = Mapper::new(config.network); move |e, p| mapper.process(e, p) }); @@ -247,6 +248,7 @@ impl Client { }; Ok(Self { + config, handle, commands: commands_rx, publisher, @@ -256,11 +258,8 @@ impl Client { /// Load the client configuration. Takes a loading handler that can optionally receive /// loading events. - pub fn load( - self, - config: Config, - loading: impl Into, - ) -> Result, Error> { + pub fn load(self, loading: impl Into) -> Result, Error> { + let config = self.config; let loading = loading.into(); let home = config.root.join(".nakamoto"); let network = config.network; @@ -397,8 +396,8 @@ impl Client { } /// Start the client process. This function is meant to be run in its own thread. - pub fn run(self, config: Config) -> Result<(), Error> { - self.load(config, LoadingHandler::Ignore)?.run() + pub fn run(self) -> Result<(), Error> { + self.load(LoadingHandler::Ignore)?.run() } /// Start the client process, supplying the service manually. diff --git a/client/src/event.rs b/client/src/event.rs index 56adef8a..f47477ba 100644 --- a/client/src/event.rs +++ b/client/src/event.rs @@ -7,6 +7,7 @@ use std::{fmt, io, net}; use nakamoto_common::bitcoin::network::constants::ServiceFlags; use nakamoto_common::bitcoin::{Transaction, Txid}; use nakamoto_common::block::{Block, BlockHash, BlockHeader, Height}; +use nakamoto_common::network::Network; use nakamoto_net::event::Emitter; use nakamoto_net::Disconnect; use nakamoto_p2p::fsm; @@ -178,6 +179,8 @@ pub enum Event { Synced { /// Height up to which we are synced. height: Height, + /// Block hash up to which we are synced. + hash: BlockHash, /// Tip of our block header chain. tip: Height, }, @@ -321,28 +324,28 @@ pub(crate) struct Mapper { sync_height: Height, /// The height up to which we've processed filters. /// This is usually going to be greater than `sync_height`. - filter_height: Height, + filter_tip: (Height, BlockHash), /// The height up to which we've processed matching blocks. /// This is always going to be lesser or equal to `filter_height`. - block_height: Height, + block_tip: (Height, BlockHash), /// Filter heights that have been matched, and for which we are awaiting a block to process. pending: HashSet, } -impl Default for Mapper { +impl Mapper { /// Create a new client event mapper. - fn default() -> Self { + pub fn new(network: Network) -> Self { let tip = 0; let sync_height = 0; - let filter_height = 0; - let block_height = 0; + let filter_tip = (0, network.genesis_hash()); + let block_tip = (0, network.genesis_hash()); let pending = HashSet::new(); Self { tip, sync_height, - filter_height, - block_height, + filter_tip, + block_tip, pending, } } @@ -439,12 +442,8 @@ impl Mapper { status: TxStatus::Acknowledged { peer }, }); } - fsm::Event::Filter(fsm::FilterEvent::RescanStarted { start, .. }) => { + fsm::Event::Filter(fsm::FilterEvent::RescanStarted { .. }) => { self.pending.clear(); - - self.filter_height = start; - self.sync_height = start; - self.block_height = start; } fsm::Event::Filter(fsm::FilterEvent::FilterProcessed { block, @@ -458,20 +457,20 @@ impl Mapper { _ => {} } assert!( - self.block_height <= self.filter_height, + self.block_tip <= self.filter_tip, "Filters are processed before blocks" ); assert!( - self.sync_height <= self.filter_height, + self.sync_height <= self.filter_tip.0, "Filters are processed before we are done" ); // If we have no blocks left to process, we are synced to the height of the last // processed filter. Otherwise, we're synced up to the last processed block. - let height = if self.pending.is_empty() { - self.filter_height.max(self.block_height) + let (height, hash) = if self.pending.is_empty() { + self.filter_tip.max(self.block_tip) } else { - self.block_height + self.block_tip }; // Ensure we only broadcast sync events when the sync height has changed. @@ -482,6 +481,7 @@ impl Mapper { emitter.emit(Event::Synced { height, + hash, tip: self.tip, }); } @@ -504,9 +504,9 @@ impl Mapper { } log::debug!("Received block {} at height {}", hash, height); - debug_assert!(height >= self.block_height); + debug_assert!(height >= self.block_tip.0); - self.block_height = height; + self.block_tip = (height, block.block_hash()); emitter.emit(Event::BlockMatched { height, @@ -526,13 +526,13 @@ impl Mapper { valid: bool, emitter: &Emitter, ) { - debug_assert!(height >= self.filter_height); + debug_assert!(height >= self.filter_tip.0); if matched { log::debug!("Filter matched for block #{}", height); self.pending.insert(height); } - self.filter_height = height; + self.filter_tip = (height, block); emitter.emit(Event::FilterProcessed { height, @@ -845,6 +845,7 @@ mod test { Event::Synced { height: sync_height, tip, + .. } => { assert_eq!(height, tip); diff --git a/node/src/lib.rs b/node/src/lib.rs index 17af4083..7822cb40 100644 --- a/node/src/lib.rs +++ b/node/src/lib.rs @@ -39,5 +39,5 @@ pub fn run( cfg.limits.max_outbound_peers = connect.len(); } - Client::::new()?.run(cfg) + Client::::new(cfg)?.run() } diff --git a/src/lib.rs b/src/lib.rs index 4a4a64a1..b1c90284 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,11 +26,11 @@ //! let cfg = Config::new(Network::Testnet); //! //! // Create a client using the above network reactor. -//! let client = Client::::new()?; +//! let client = Client::::new(cfg)?; //! let handle = client.handle(); //! //! // Run the client on a different thread, to not block the main thread. -//! thread::spawn(|| client.run(cfg).unwrap()); +//! thread::spawn(|| client.run().unwrap()); //! //! // Wait for the client to be connected to a peer. //! handle.wait_for_peers(1, Services::default())?; diff --git a/wallet/src/lib.rs b/wallet/src/lib.rs index 390f9dea..f9b9a0b0 100644 --- a/wallet/src/lib.rs +++ b/wallet/src/lib.rs @@ -43,7 +43,7 @@ pub fn run( }; // Create a new client using `Reactor` for networking. - let client = Client::::new()?; + let client = Client::::new(cfg)?; let handle = client.handle(); let client_recv = handle.events(); let (loading_send, loading_recv) = chan::unbounded(); @@ -68,7 +68,7 @@ pub fn run( if offline { Ok(()) } else { - client.load(cfg, loading_send)?.run() + client.load(loading_send)?.run() } }); diff --git a/wallet/src/wallet.rs b/wallet/src/wallet.rs index 13f482d7..409470ef 100644 --- a/wallet/src/wallet.rs +++ b/wallet/src/wallet.rs @@ -259,7 +259,7 @@ impl Wallet { ); } // TODO: This should be called `Scanned`. - client::Event::Synced { height, tip } => { + client::Event::Synced { height, tip, .. } => { self.ui.handle_synced(height, tip); } _ => {} From 5bdfbc4d86a801b79e42be7e947eeec75289793f Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sat, 3 Dec 2022 00:42:57 +0100 Subject: [PATCH 3/8] [wallet] Keep track of last sync height --- wallet/src/wallet.rs | 35 +++++++++++++++++++++++---- wallet/src/wallet/db.rs | 46 ++++++++++++++++++++++++++++++++++++ wallet/src/wallet/schema.sql | 6 +++++ 3 files changed, 82 insertions(+), 5 deletions(-) diff --git a/wallet/src/wallet.rs b/wallet/src/wallet.rs index 409470ef..5603d592 100644 --- a/wallet/src/wallet.rs +++ b/wallet/src/wallet.rs @@ -128,9 +128,6 @@ impl Wallet { if offline { ui::refresh(&mut self.ui, &self.db, &mut term)?; } else { - // Start a re-scan from the birht height, which keeps scanning as new blocks arrive. - self.client.rescan(birth.., watch.iter().cloned())?; - // Loading... loop { chan::select! { @@ -182,7 +179,9 @@ impl Wallet { recv(events) -> event => { let event = event?; - if let Break(()) = self.handle_client_event(event, &watch, offline, &mut term)? { + if let Break(()) = self.handle_client_event( + event, &watch, offline, birth, &mut term + )? { break; } } @@ -226,6 +225,7 @@ impl Wallet { event: client::Event, watch: &[Script], offline: bool, + birth: Height, term: &mut W, ) -> Result, Error> { log::debug!("Received event: {}", event); @@ -233,6 +233,28 @@ impl Wallet { match event { client::Event::Ready { tip, .. } => { self.ui.handle_ready(tip, offline); + + let start = if let Some((sync_height, sync_hash)) = self.db.sync_state()? { + log::info!("Previous sync state found with height {sync_height}.."); + + if let Some(header) = self.client.get_block_by_height(sync_height)? { + if header.block_hash() == sync_hash { + sync_height + } else { + log::warn!("Previous sync tip {sync_hash} at height {sync_height} is no longer on the active chain"); + birth + } + } else { + log::warn!("Previous sync height {sync_height} was not found, reverting to current tip"); + tip + } + } else { + log::info!("No previous sync state, starting from birth ({birth})"); + birth + }; + + // Start a re-scan from the birht height, which keeps scanning as new blocks arrive. + self.client.rescan(start.., watch.iter().cloned())?; } client::Event::PeerHeightUpdated { height } => { self.ui.handle_peer_height(height); @@ -259,8 +281,11 @@ impl Wallet { ); } // TODO: This should be called `Scanned`. - client::Event::Synced { height, tip, .. } => { + client::Event::Synced { height, hash, tip } => { self.ui.handle_synced(height, tip); + self.db.synced_to(height, hash)?; + + log::info!("Synced up to height {height}"); } _ => {} } diff --git a/wallet/src/wallet/db.rs b/wallet/src/wallet/db.rs index e61ea012..31a560c9 100644 --- a/wallet/src/wallet/db.rs +++ b/wallet/src/wallet/db.rs @@ -7,6 +7,7 @@ use nakamoto_common::bitcoin::Address; use nakamoto_common::bitcoin::OutPoint; use nakamoto_common::bitcoin::TxOut; use nakamoto_common::bitcoin::Txid; +use nakamoto_common::block::{BlockHash, Height}; use sqlite as sql; @@ -36,6 +37,8 @@ pub trait Read { fn utxos(&self) -> Result, Error>; /// Get all addresses. fn addresses(&self) -> Result, Error>; + /// Get sync state. + fn sync_state(&self) -> Result, Error>; } /// Write to the database. @@ -51,6 +54,8 @@ pub trait Write { index: usize, label: Option<&str>, ) -> Result; + /// Save height up to which we're synced. + fn synced_to(&self, height: Height, hash: BlockHash) -> Result; } /// Wallet database. @@ -145,6 +150,28 @@ impl Read for Db { } Ok(addrs) } + + fn sync_state(&self) -> Result, Error> { + let row = self + .raw + .prepare( + "SELECT `height`, `hash` + FROM sync + LIMIT 1", + )? + .into_cursor() + .next(); + + let Some(Ok(row)) = row else { + return Ok(None); + }; + + let height = row.get::("height") as Height; + let hash = row.get::("hash"); + let hash = BlockHash::from_str(&hash).unwrap(); + + Ok(Some((height, hash))) + } } impl Write for Db { @@ -207,6 +234,25 @@ impl Write for Db { Ok(self.raw.change_count() > 0) } + + fn synced_to(&self, height: Height, hash: BlockHash) -> Result { + self.raw + .prepare( + "INSERT INTO sync (`network`, `height`, `hash`) + VALUES (?1, ?2, ?3) + ON CONFLICT DO UPDATE + SET height = ?2, hash = ?3", + )? + .into_cursor() + .bind(&[ + sql::Value::String(String::from("bitcoin")), + sql::Value::Integer(height as i64), + sql::Value::String(hash.to_string()), + ])? + .next(); + + Ok(self.raw.change_count() > 0) + } } impl Db { diff --git a/wallet/src/wallet/schema.sql b/wallet/src/wallet/schema.sql index a055d53f..b8658237 100644 --- a/wallet/src/wallet/schema.sql +++ b/wallet/src/wallet/schema.sql @@ -16,3 +16,9 @@ CREATE TABLE IF NOT EXISTS "addresses" ( "received" integer NOT NULL DEFAULT 0, "used" integer NOT NULL DEFAULT false ) STRICT; + +CREATE TABLE IF NOT EXISTS "sync" ( + "network" text NOT NULL UNIQUE, + "height" integer NOT NULL UNIQUE, + "hash" text NOT NULL UNIQUE +) STRICT; From 904ef5707ca3127b1f4094672bd12248c7f8bc6d Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sat, 3 Dec 2022 00:43:51 +0100 Subject: [PATCH 4/8] [wallet] Add `Ui::set_message` function --- wallet/src/wallet/ui.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/wallet/src/wallet/ui.rs b/wallet/src/wallet/ui.rs index dba67d0c..1743d9d2 100644 --- a/wallet/src/wallet/ui.rs +++ b/wallet/src/wallet/ui.rs @@ -177,6 +177,11 @@ impl Ui { ) } + pub fn set_message(&mut self, message: impl Into) { + self.message = message.into(); + self.redraw |= REDRAW_FOOTER; + } + pub fn set_balance(&mut self, balance: u64) { self.balance = Balance(balance); self.redraw |= REDRAW_HEADER; @@ -489,8 +494,9 @@ pub fn draw_footer(ui: &Ui, term: &mut W) -> io::Result<()> { write!( term, - "{}{}{}{}", + "{}{}{}{}{}", cursor::Goto(1, height - 1), + clear::CurrentLine, color::Bg(color::Reset), color::Fg(color::Blue), ui.message, From 6cd86cc3187fc1cc9d33908692b74d05c33b53db Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sat, 3 Dec 2022 00:44:23 +0100 Subject: [PATCH 5/8] [wallet] Show connected peer count --- wallet/src/wallet.rs | 29 +++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/wallet/src/wallet.rs b/wallet/src/wallet.rs index 5603d592..eb4ded01 100644 --- a/wallet/src/wallet.rs +++ b/wallet/src/wallet.rs @@ -3,10 +3,11 @@ pub mod hw; pub mod ui; use std::collections::HashSet; -use std::io; use std::ops::ControlFlow; use std::ops::ControlFlow::*; +use std::{io, net}; +use client::Link; use crossbeam_channel as chan; use termion::event::Event; @@ -33,6 +34,7 @@ pub struct Wallet { ui: Ui, hw: Hw, network: client::Network, + peers: HashSet, watch: HashSet
, } @@ -45,6 +47,7 @@ impl Wallet { hw, network, watch: HashSet::new(), + peers: HashSet::new(), ui: Ui::default(), } } @@ -119,7 +122,11 @@ impl Wallet { let watch: Vec<_> = self.watch.iter().map(|a| a.script_pubkey()).collect(); let balance = self.db.balance()?; - self.ui.message = format!("Scanning from block height {}", birth); + self.ui.message = if offline { + String::from("Offline") + } else { + String::from("Loading...") + }; self.ui.reset(&mut term)?; self.ui.decorations(&mut term)?; self.ui.set_balance(balance); @@ -157,6 +164,7 @@ impl Wallet { } ui::refresh(&mut self.ui, &self.db, &mut term)?; } + self.ui.set_message("Ready"); } // Running... @@ -255,6 +263,23 @@ impl Wallet { // Start a re-scan from the birht height, which keeps scanning as new blocks arrive. self.client.rescan(start.., watch.iter().cloned())?; + self.ui + .set_message(format!("Scanning from block height {}", start)); + } + client::Event::PeerNegotiated { + addr, + link: Link::Outbound, + .. + } => { + self.peers.insert(addr); + self.ui + .set_message(format!("Connected to {} peer(s)", self.peers.len())); + } + client::Event::PeerDisconnected { addr, .. } => { + if self.peers.remove(&addr) { + self.ui + .set_message(format!("Connected to {} peer(s)", self.peers.len())); + } } client::Event::PeerHeightUpdated { height } => { self.ui.handle_peer_height(height); From 920d6be3020fed482f2346e1f1ac9f26ca19afcf Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sat, 3 Dec 2022 00:44:39 +0100 Subject: [PATCH 6/8] [wallet] Set 'info' logging by default --- wallet/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/wallet/src/main.rs b/wallet/src/main.rs index 23f4d198..0e19bd91 100644 --- a/wallet/src/main.rs +++ b/wallet/src/main.rs @@ -50,7 +50,7 @@ fn main() { let level = if opts.debug { log::Level::Debug } else { - log::Level::Error + log::Level::Info }; logger::init(level).expect("initializing logger for the first time"); From 0add124ff16743560cd9b8e2946e0b640b5d7bce Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Sat, 3 Dec 2022 11:01:13 +0100 Subject: [PATCH 7/8] [client] Add hash in ready event --- client/src/event.rs | 4 ++++ p2p/src/fsm.rs | 1 + p2p/src/fsm/event.rs | 3 +++ 3 files changed, 8 insertions(+) diff --git a/client/src/event.rs b/client/src/event.rs index f47477ba..fc4b8e94 100644 --- a/client/src/event.rs +++ b/client/src/event.rs @@ -61,6 +61,8 @@ pub enum Event { Ready { /// The tip of the block header chain. tip: Height, + /// The hash of the tip. + hash: BlockHash, /// The tip of the filter header chain. filter_tip: Height, }, @@ -357,11 +359,13 @@ impl Mapper { match event { fsm::Event::Ready { height, + hash, filter_height, .. } => { emitter.emit(Event::Ready { tip: height, + hash, filter_tip: filter_height, }); } diff --git a/p2p/src/fsm.rs b/p2p/src/fsm.rs index 56a9bfa7..606e7182 100644 --- a/p2p/src/fsm.rs +++ b/p2p/src/fsm.rs @@ -782,6 +782,7 @@ impl> traits: self.cbfmgr.initialize(&self.tree); self.outbox.event(Event::Ready { height: self.tree.height(), + hash: self.tree.tip().0, filter_height: self.cbfmgr.filters.height(), time, }); diff --git a/p2p/src/fsm/event.rs b/p2p/src/fsm/event.rs index d2e1a77f..67cf9ba7 100644 --- a/p2p/src/fsm/event.rs +++ b/p2p/src/fsm/event.rs @@ -1,5 +1,6 @@ //! State machine events. use nakamoto_common::bitcoin::network::message::NetworkMessage; +use nakamoto_common::bitcoin::BlockHash; use crate::fsm::{self, Height, LocalTime, PeerId}; @@ -12,6 +13,8 @@ pub enum Event { Ready { /// Block header height. height: Height, + /// Block header hash. + hash: BlockHash, /// Filter header height. filter_height: Height, /// Local time. From 1f1ac95f1e1d8041750180de922c69afe1b4ab9d Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Mon, 2 Jan 2023 22:05:43 +0100 Subject: [PATCH 8/8] WIP --- client/src/client.rs | 5 +++-- client/src/handle.rs | 43 +++++++++++++++++++++++++++++++++++++++---- wallet/src/wallet.rs | 2 +- 3 files changed, 43 insertions(+), 7 deletions(-) diff --git a/client/src/client.rs b/client/src/client.rs index 68fc8074..0b5654ef 100644 --- a/client/src/client.rs +++ b/client/src/client.rs @@ -39,6 +39,7 @@ pub use nakamoto_p2p::fsm::{Command, CommandError, Hooks, Limits, Link, Peer}; pub use crate::error::Error; pub use crate::event::{Event, Loading}; pub use crate::handle; +pub use crate::handle::Events; pub use crate::service::Service; use crate::event::Mapper; @@ -551,8 +552,8 @@ impl handle::Handle for Handle { self.filters.subscribe() } - fn events(&self) -> chan::Receiver { - self.subscriber.subscribe() + fn events(&self) -> handle::Events { + handle::Events::from(self.subscriber.subscribe()) } fn command(&self, cmd: Command) -> Result<(), handle::Error> { diff --git a/client/src/handle.rs b/client/src/handle.rs index a0728008..d1758349 100644 --- a/client/src/handle.rs +++ b/client/src/handle.rs @@ -1,21 +1,21 @@ //! Node handles are created from nodes by users of the library, to communicate with the underlying //! protocol instance. use std::net; -use std::ops::{RangeBounds, RangeInclusive}; +use std::ops::{Deref, RangeBounds, RangeInclusive}; use crossbeam_channel as chan; use thiserror::Error; use nakamoto_common::bitcoin::network::constants::ServiceFlags; +use nakamoto_common::bitcoin::network::message::NetworkMessage; use nakamoto_common::bitcoin::network::Address; use nakamoto_common::bitcoin::util::uint::Uint256; use nakamoto_common::bitcoin::Script; - -use nakamoto_common::bitcoin::network::message::NetworkMessage; use nakamoto_common::block::filter::BlockFilter; use nakamoto_common::block::tree::{BlockReader, ImportResult}; use nakamoto_common::block::{self, Block, BlockHash, BlockHeader, Height, Transaction}; use nakamoto_common::nonempty::NonEmpty; +use nakamoto_net::event; use nakamoto_p2p::fsm::Link; use nakamoto_p2p::fsm::{self, Command, CommandError, GetFiltersError, Peer}; @@ -62,6 +62,41 @@ impl From> for Error { } } +/// [`Event`] receiver. +#[derive(Debug, Clone)] +pub struct Events { + receiver: chan::Receiver, +} + +impl Events { + /// Wait for the readiness event. + pub fn wait_for_ready(&self) -> Result<(Height, BlockHash), Error> { + event::wait( + &self.receiver, + |event| match event { + Event::Ready { tip, hash, .. } => Some((tip, hash)), + _ => None, + }, + std::time::Duration::MAX, + ) + .map_err(Error::from) + } +} + +impl From> for Events { + fn from(receiver: chan::Receiver) -> Self { + Self { receiver } + } +} + +impl Deref for Events { + type Target = chan::Receiver; + + fn deref(&self) -> &Self::Target { + &self.receiver + } +} + /// A handle for communicating with a node process. pub trait Handle: Sized + Send + Sync + Clone { /// Get the tip of the active chain. Returns the height of the chain, the header, @@ -95,7 +130,7 @@ pub trait Handle: Sized + Send + Sync + Clone { /// Subscribe to compact filters received. fn filters(&self) -> chan::Receiver<(BlockFilter, BlockHash, Height)>; /// Subscribe to client events. - fn events(&self) -> chan::Receiver; + fn events(&self) -> Events; /// Send a command to the client. fn command(&self, cmd: Command) -> Result<(), Error>; diff --git a/wallet/src/wallet.rs b/wallet/src/wallet.rs index eb4ded01..b989bf30 100644 --- a/wallet/src/wallet.rs +++ b/wallet/src/wallet.rs @@ -91,7 +91,7 @@ impl Wallet { inputs: chan::Receiver, signals: chan::Receiver, loading: chan::Receiver, - events: chan::Receiver, + events: client::Events, offline: bool, mut term: W, ) -> Result<(), Error> {