Wallet stuff

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;
@@ -189,14 +190,15 @@ impl<R: Reactor> ClientRunner<R> {
 /// A light-client process.
 pub struct Client<R: Reactor> {
     handle: Handle<R::Waker>,
+    config: Config,
     commands: chan::Receiver<Command>,
     publisher: Publisher<fsm::Event>,
     reactor: R,
 }
 
 impl<R: Reactor> Client<R> {
     /// Create a new client.
-    pub fn new() -> Result<Self, Error> {
+    pub fn new(config: Config) -> Result<Self, Error> {
         let (commands_tx, commands_rx) = chan::unbounded::<Command>();
         let (event_pub, events) = event::broadcast(|e, p| p.emit(e));
         let (blocks_pub, blocks) = event::broadcast(|e, p| {
@@ -221,7 +223,7 @@ impl<R: Reactor> Client<R> {
             }
         });
         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 +249,7 @@ impl<R: Reactor> Client<R> {
         };
 
         Ok(Self {
+            config,
             handle,
             commands: commands_rx,
             publisher,
@@ -256,11 +259,8 @@ impl<R: Reactor> Client<R> {
 
     /// Load the client configuration. Takes a loading handler that can optionally receive
     /// loading events.
-    pub fn load(
-        self,
-        config: Config,
-        loading: impl Into<LoadingHandler>,
-    ) -> Result<ClientRunner<R>, Error> {
+    pub fn load(self, loading: impl Into<LoadingHandler>) -> Result<ClientRunner<R>, Error> {
+        let config = self.config;
         let loading = loading.into();
         let home = config.root.join(".nakamoto");
         let network = config.network;
@@ -397,8 +397,8 @@ impl<R: Reactor> Client<R> {
     }
 
     /// 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.
@@ -552,8 +552,8 @@ impl<W: Waker> handle::Handle for Handle<W> {
         self.filters.subscribe()
     }
 
-    fn events(&self) -> chan::Receiver<Event> {
-        self.subscriber.subscribe()
+    fn events(&self) -> handle::Events {
+        handle::Events::from(self.subscriber.subscribe())
     }
 
     fn command(&self, cmd: Command) -> Result<(), handle::Error> {

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;
@@ -60,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,
     },
@@ -178,6 +181,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 +326,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<Height>,
 }
 
-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,
         }
     }
@@ -354,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,
                 });
             }
@@ -439,12 +446,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,28 +461,31 @@ 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
+        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.
         if height > self.sync_height {
+            debug_assert!(height == self.sync_height + 1);
+
             self.sync_height = height;
 
             emitter.emit(Event::Synced {
                 height,
+                hash,
                 tip: self.tip,
             });
         }
@@ -502,9 +508,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,
@@ -524,13 +530,13 @@ impl Mapper {
         valid: bool,
         emitter: &Emitter<Event>,
     ) {
-        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,
@@ -843,6 +849,7 @@ mod test {
                 Event::Synced {
                     height: sync_height,
                     tip,
+                    ..
                 } => {
                     assert_eq!(height, tip);
 

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<T> From<chan::SendError<T>> for Error {
     }
 }
 
+/// [`Event`] receiver.
+#[derive(Debug, Clone)]
+pub struct Events {
+    receiver: chan::Receiver<Event>,
+}
+
+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<chan::Receiver<Event>> for Events {
+    fn from(receiver: chan::Receiver<Event>) -> Self {
+        Self { receiver }
+    }
+}
+
+impl Deref for Events {
+    type Target = chan::Receiver<Event>;
+
+    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<Event>;
+    fn events(&self) -> Events;
 
     /// Send a command to the client.
     fn command(&self, cmd: Command) -> Result<(), Error>;

net/poll/src/reactor.rs

@@ -305,7 +305,7 @@ impl<Id: PeerId> Reactor<net::TcpStream, Id> {
 
                     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<Id: PeerId> Reactor<net::TcpStream, Id> {
                     self.timeouts.register((), local_time + timeout);
                 }
                 Io::Event(event) => {
-                    trace!("Event: {:?}", event);
-
                     publisher.publish(event);
                 }
             }

node/src/lib.rs

@@ -39,5 +39,5 @@ pub fn run(
         cfg.limits.max_outbound_peers = connect.len();
     }
 
-    Client::<Reactor>::new()?.run(cfg)
+    Client::<Reactor>::new(cfg)?.run()
 }

p2p/src/fsm.rs

@@ -782,6 +782,7 @@ impl<T: BlockTree, F: Filters, P: peer::Store, C: AdjustedClock<PeerId>> 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,
         });

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<F: Filters, U: Wire<Event> + 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<F: Filters, U: Wire<Event> + 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<F: Filters, U: Wire<Event> + 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
         );