Enforce client semantics with state

Uses the sealed trait hack to only allow for compiler enforced state
transitions. The flow is `Idle -> Subscribed -> Actice`, where a user
decides what to do with the events, runs the node, then gets access to
all the usual client features.

I decided to scrap the `Requester` and instead encapsulate the event
receivers with a newtype.

The only downside I see with this is `&mut self` doesn't seem to be
possible, as each impl block is defined for it's `State` type. I need to
fiddle with that to see if we can't remove the need to return the client
each state transition.

Author: rustaceanrob

examples/bitcoin.rs

@@ -3,7 +3,8 @@
 
 use bip157::builder::Builder;
 use bip157::chain::{BlockHeaderChanges, ChainState};
-use bip157::{lookup_host, Client, Event, HeaderCheckpoint, Network, ScriptBuf};
+use bip157::client::EventListeners;
+use bip157::{lookup_host, Event, HeaderCheckpoint, Network, ScriptBuf};
 use std::collections::HashSet;
 use tokio::time::Instant;
 
@@ -35,17 +36,17 @@ async fn main() {
         // Create the node and client
         .build();
 
-    let client = client.run();
     // Split the client into components that send messages and listen to messages.
     // With this construction, different parts of the program can take ownership of
     // specific tasks.
-    let Client {
-        requester,
+    let (client, events) = client.subscribe();
+    let EventListeners {
         mut info_rx,
         mut warn_rx,
         mut event_rx,
-        ..
-    } = client;
+    } = events;
+    let client = client.start();
+
     // Continually listen for events until the node is synced to its peers.
     loop {
         tokio::select! {
@@ -55,11 +56,11 @@ async fn main() {
                         Event::FiltersSynced(update) => {
                             tracing::info!("Chain tip: {}",update.tip().hash);
                             // Request information from the node
-                            let fee = requester.broadcast_min_feerate().await.unwrap();
+                            let fee = client.broadcast_min_feerate().await.unwrap();
                             tracing::info!("Minimum transaction broadcast fee rate: {:#}", fee);
                             let sync_time = now.elapsed().as_secs_f32();
                             tracing::info!("Total sync time: {sync_time} seconds");
-                            let avg_fee_rate = requester.average_fee_rate(update.tip().hash).await.unwrap();
+                            let avg_fee_rate = client.average_fee_rate(update.tip().hash).await.unwrap();
                             tracing::info!("Last block average fee rate: {:#}", avg_fee_rate);
                             break;
                         },
@@ -82,6 +83,6 @@ async fn main() {
             }
         }
     }
-    let _ = requester.shutdown();
+    let _ = client.shutdown();
     tracing::info!("Shutting down");
 }

examples/signet.rs

@@ -3,8 +3,8 @@
 
 use bip157::chain::{BlockHeaderChanges, ChainState};
 use bip157::messages::Event;
-use bip157::{builder::Builder, chain::checkpoints::HeaderCheckpoint, Client};
-use bip157::{Address, BlockHash, Network};
+use bip157::{builder::Builder, chain::checkpoints::HeaderCheckpoint};
+use bip157::{Address, BlockHash, EventListeners, Network};
 use std::collections::HashSet;
 use std::str::FromStr;
 
@@ -40,15 +40,13 @@ async fn main() {
         // Create the node and client
         .build();
 
-    let client = client.run();
-
-    let Client {
-        requester,
+    let (client, events) = client.subscribe();
+    let EventListeners {
         mut info_rx,
         mut warn_rx,
         mut event_rx,
-        ..
-    } = client;
+    } = events;
+    let client = client.start();
 
     // Continually listen for events until the node is synced to its peers.
     loop {
@@ -72,7 +70,7 @@ async fn main() {
                             if filter.contains_any(addresses.iter()) {
                                 let hash = filter.block_hash();
                                 tracing::info!("Found script at {}!", hash);
-                                let indexed_block = requester.get_block(hash).await.unwrap();
+                                let indexed_block = client.get_block(hash).await.unwrap();
                                 let coinbase = indexed_block.block.txdata.first().unwrap().compute_txid();
                                 tracing::info!("Coinbase transaction ID: {}", coinbase);
                                 break;
@@ -88,6 +86,6 @@ async fn main() {
             }
         }
     }
-    let _ = requester.shutdown();
+    let _ = client.shutdown();
     tracing::info!("Shutting down");
 }

src/builder.rs

@@ -5,6 +5,7 @@ use bitcoin::Network;
 
 use super::{client::Client, node::Node};
 use crate::chain::ChainState;
+use crate::client::Idle;
 use crate::network::ConnectionType;
 use crate::TrustedPeer;
 use crate::{Config, FilterType};
@@ -139,7 +140,7 @@ impl Builder {
     }
 
     /// Consume the node builder and receive a [`Client`].
-    pub fn build(mut self) -> Client {
+    pub fn build(mut self) -> Client<Idle> {
         Node::build(self.network, core::mem::take(&mut self.config))
     }
 }

src/client.rs

@@ -12,58 +12,135 @@ use crate::{Event, Info, TrustedPeer, Warning};
 use super::{error::ClientError, messages::ClientMessage};
 use super::{error::FetchBlockError, IndexedBlock};
 
-/// A [`Client`] allows for communication with a running node.
+/// Client state when idle.
+pub struct Idle;
+/// Client state when subscribed to events.
+pub struct Subscribed;
+/// Client state when active.
+pub struct Active;
+
+mod sealed {
+    pub trait Sealed {}
+}
+
+impl sealed::Sealed for Idle {}
+impl sealed::Sealed for Subscribed {}
+impl sealed::Sealed for Active {}
+
+/// State of the client.
+pub trait State: sealed::Sealed {}
+
+impl State for Idle {}
+impl State for Subscribed {}
+impl State for Active {}
+
+/// Wrapper type for the channels that will receive events.
 #[derive(Debug)]
-pub struct Client {
-    /// Send events to a node, such as broadcasting a transaction.
-    pub requester: Requester,
+pub struct EventListeners {
     /// Receive informational messages from the node.
     pub info_rx: mpsc::Receiver<Info>,
     /// Receive warning messages from a node.
     pub warn_rx: mpsc::UnboundedReceiver<Warning>,
     /// Receive [`Event`] from a node to act on.
     pub event_rx: mpsc::UnboundedReceiver<Event>,
-    /// Internal node structure.
-    node: Option<Node>,
 }
 
-impl Client {
-    pub(crate) fn new(
+impl EventListeners {
+    fn new(
         info_rx: mpsc::Receiver<Info>,
         warn_rx: mpsc::UnboundedReceiver<Warning>,
         event_rx: mpsc::UnboundedReceiver<Event>,
-        ntx: UnboundedSender<ClientMessage>,
-        node: Node,
     ) -> Self {
         Self {
-            requester: Requester::new(ntx),
             info_rx,
             warn_rx,
             event_rx,
+        }
+    }
+}
+
+/// A [`Client`] allows for communication with a running node.
+///
+/// The [`Client`] is generic over 3 states:
+/// - [`Idle`]: the client is not running and event handling has not been initialized.
+/// - [`Subscribed`]: events have been subscribed to in the program, but the client has not started.
+/// - [`Active`]: data is actively being fetched and the [`Client`] may perform actions.
+///
+#[derive(Debug)]
+pub struct Client<S: State> {
+    /// Send events to a node, such as broadcasting a transaction.
+    ntx: UnboundedSender<ClientMessage>,
+    /// Receive informational messages from the node.
+    events: Option<EventListeners>,
+    /// Internal node structure.
+    node: Option<Node>,
+    /// Marker for state.
+    _marker: core::marker::PhantomData<S>,
+}
+
+impl Client<Idle> {
+    pub(crate) fn new(
+        info_rx: mpsc::Receiver<Info>,
+        warn_rx: mpsc::UnboundedReceiver<Warning>,
+        event_rx: mpsc::UnboundedReceiver<Event>,
+        ntx: UnboundedSender<ClientMessage>,
+        node: Node,
+    ) -> Client<Idle> {
+        Client {
+            ntx,
+            events: Some(EventListeners::new(info_rx, warn_rx, event_rx)),
             node: Some(node),
+            _marker: core::marker::PhantomData,
         }
     }
 
-    /// Start the underlying node on a [`tokio::task`]. This assumes there is a runtime present to
-    /// execute the task.
-    pub fn run(mut self) -> Self {
-        let node = core::mem::take(&mut self.node).expect("cannot call run twice.");
-        tokio::task::spawn(async move { node.run().await });
-        self
+    /// Subscribe to the events published by the light client. Applications may perform arbitrary behavior
+    /// when receiving these events, such as logging or applying the effect of a block to a wallet.
+    /// The client is not yet running after this step.
+    pub fn subscribe(mut self) -> (Client<Subscribed>, EventListeners) {
+        let events = core::mem::take(&mut self.events).expect("cannot call run twice.");
+        (
+            Client {
+                ntx: self.ntx,
+                events: None,
+                node: self.node,
+                _marker: core::marker::PhantomData,
+            },
+            events,
+        )
     }
 }
 
-/// Send messages to a node that is running so the node may complete a task.
-#[derive(Debug, Clone)]
-pub struct Requester {
-    ntx: UnboundedSender<ClientMessage>,
-}
+impl Client<Subscribed> {
+    /// Start the client, which will begin publishing events to subscribers. This will implicitly
+    /// spawn a [`tokio::task`] to fetch data for the client.
+    pub fn start(mut self) -> Client<Active> {
+        let node = core::mem::take(&mut self.node).expect("cannot call run twice.");
+        tokio::task::spawn(async move { node.run().await });
+        Client {
+            ntx: self.ntx,
+            events: None,
+            node: None,
+            _marker: core::marker::PhantomData,
+        }
+    }
 
-impl Requester {
-    fn new(ntx: UnboundedSender<ClientMessage>) -> Self {
-        Self { ntx }
+    /// Receive a [`Node`] to run on a dedicated resource, likely with a custom [`tokio::runtime::Runtime`].
+    pub fn start_managed(mut self) -> (Client<Active>, Node) {
+        let node = core::mem::take(&mut self.node).expect("cannot call run twice.");
+        (
+            Client {
+                ntx: self.ntx,
+                events: None,
+                node: None,
+                _marker: core::marker::PhantomData,
+            },
+            node,
+        )
     }
+}
 
+impl Client<Active> {
     /// Tell the node to shut down.
     ///
     /// # Errors

src/lib.rs

@@ -8,7 +8,7 @@
 //! # Example usage
 //!
 //! ```no_run
-//! use bip157::{Builder, Event, Client, Network, BlockHash};
+//! use bip157::{Builder, Event, EventListeners, Client, Network, BlockHash};
 //!
 //! #[tokio::main]
 //! async fn main() {
@@ -22,10 +22,12 @@
 //!         // The number of connections we would like to maintain
 //!         .required_peers(2)
 //!         .build();
-//!     // Run the node and wait for the sync message;
-//!     let client = client.run();
+//!     // Start the node
+//!     let (client, events) = client.subscribe();
+//!     let client = client.start();
 //!     // Split the client into components that send messages and listen to messages
-//!     let Client { requester, info_rx: _, warn_rx: _, mut event_rx, .. } = client;
+//!     let EventListeners { info_rx: _, warn_rx: _, mut event_rx } = events;
+//!     // Wait for the sync message;
 //!     loop {
 //!         if let Some(event) = event_rx.recv().await {
 //!             match event {
@@ -37,7 +39,7 @@
 //!             }
 //!         }
 //!     }
-//!     requester.shutdown();
+//!     client.shutdown();
 //! }
 //! ```
 
@@ -81,7 +83,7 @@ use tokio::sync::mpsc::UnboundedSender;
 pub use {
     crate::builder::Builder,
     crate::chain::ChainState,
-    crate::client::{Client, Requester},
+    crate::client::{Client, EventListeners},
     crate::error::{ClientError, NodeError},
     crate::messages::{Event, Info, Progress, RejectPayload, SyncUpdate, Warning},
     crate::node::Node,

src/node.rs

@@ -31,6 +31,7 @@ use crate::{
         error::HeaderSyncError,
         CFHeaderChanges, ChainState, FilterCheck, HeightMonitor,
     },
+    client::Idle,
     error::FetchBlockError,
     messages::ClientRequest,
     network::{
@@ -66,7 +67,7 @@ pub struct Node {
 }
 
 impl Node {
-    pub(crate) fn build(network: Network, config: Config) -> Client {
+    pub(crate) fn build(network: Network, config: Config) -> Client<Idle> {
         let Config {
             required_peers,
             white_list,