bark/daemon/tip_watcher/
mod.rs

1//! Watch blockchain tip changes.
2
3mod polling;
4mod source;
5#[cfg(all(feature = "bitcoind-rpc", not(target_arch = "wasm32")))]
6mod zmq;
7
8use std::sync::Arc;
9use std::time::Duration;
10
11use tokio::sync::watch;
12
13use bark_runtime::CancellationToken;
14use bitcoin_ext::BlockRef;
15
16pub use source::TipSource;
17
18use polling::PollingTipWatcher;
19#[cfg(all(feature = "bitcoind-rpc", not(target_arch = "wasm32")))]
20use zmq::ZmqTipWatcher;
21
22#[derive(Clone)]
23pub struct TipWatcher {
24	shutdown: CancellationToken,
25	/// Stops the watcher task once the last clone is dropped.
26	_shutdown_on_drop: Arc<ShutdownGuard>,
27	rx: watch::Receiver<BlockRef>,
28}
29
30struct ShutdownGuard(CancellationToken);
31
32impl Drop for ShutdownGuard {
33	fn drop(&mut self) {
34		self.0.cancel();
35	}
36}
37
38impl TipWatcher {
39	fn new(shutdown: CancellationToken, rx: watch::Receiver<BlockRef>) -> Self {
40		let guard = ShutdownGuard(shutdown.clone());
41		Self { shutdown, _shutdown_on_drop: Arc::new(guard), rx }
42	}
43
44	pub async fn start_poll<S: TipSource + 'static>(
45		source: Arc<S>,
46		poll_interval: Duration,
47	) -> anyhow::Result<Self> {
48		let initial = source.tip_ref().await?;
49		let (tx, rx) = watch::channel(initial);
50		let shutdown = CancellationToken::new();
51		let proc = PollingTipWatcher {
52			source,
53			poll_interval,
54			shutdown: shutdown.clone(),
55			tx,
56		};
57
58		bark_runtime::spawn(proc.run());
59		Ok(Self::new(shutdown, rx))
60	}
61
62	/// Start a watcher that follows the chain tip of `source`.
63	///
64	/// The watcher has two sources of tip updates and trusts both.
65	///
66	/// bitcoind announces the hash of each new block on `zmq_endpoint`.
67	/// The watcher makes this hash the new tip and increases the height
68	/// by one. No backend call is necessary for this update.
69	///
70	/// After each `reconcile_interval`, the watcher reads the tip from
71	/// `source`. A changed value becomes the new tip. This read corrects
72	/// the tip after a lost notification or a reorg.
73	#[cfg(all(feature = "bitcoind-rpc", not(target_arch = "wasm32")))]
74	pub async fn start_zmq<S: TipSource + 'static>(
75		source: Arc<S>,
76		zmq_endpoint: &str,
77		reconcile_interval: Duration,
78	) -> anyhow::Result<Self> {
79		let socket = zmq::connect(zmq_endpoint).await?;
80		let initial = source.tip_ref().await?;
81		let (tx, rx) = watch::channel(initial);
82		let shutdown = CancellationToken::new();
83
84		let proc = ZmqTipWatcher {
85			source,
86			reconcile_interval,
87			shutdown: shutdown.clone(),
88			tx,
89			socket,
90		};
91
92		bark_runtime::spawn(proc.run());
93		Ok(Self::new(shutdown, rx))
94	}
95
96	pub fn tip(&self) -> BlockRef {
97		*self.rx.borrow()
98	}
99
100	/// Subscribe to tip changes.
101	///
102	/// The subscription keeps the watcher task alive.
103	pub fn subscribe(&self) -> TipSubscription {
104		TipSubscription {
105			rx: self.rx.clone(),
106			_shutdown_on_drop: self._shutdown_on_drop.clone(),
107		}
108	}
109
110	/// Wait until the tip reaches the given height.
111	pub async fn wait_for_height(&self, height: u32) -> anyhow::Result<BlockRef> {
112		// watch::Receiver::wait_for needs a mutable receiver,
113		// so we subscribe with our own copy
114		let mut subscription = self.rx.clone();
115		let tip = subscription.wait_for(|tip| tip.height >= height).await?;
116		Ok(*tip)
117	}
118
119	pub fn stop(&self) {
120		self.shutdown.cancel();
121	}
122}
123
124/// A subscription to tip changes handed out by [TipWatcher::subscribe].
125///
126/// Keeps the watcher task alive for as long as it is held.
127#[derive(Clone)]
128pub struct TipSubscription {
129	rx: watch::Receiver<BlockRef>,
130	_shutdown_on_drop: Arc<ShutdownGuard>,
131}
132
133impl TipSubscription {
134	/// The current tip.
135	pub fn tip(&self) -> BlockRef {
136		*self.rx.borrow()
137	}
138
139	/// Wait for the tip to change and return the new tip.
140	///
141	/// Errors when the watcher task has stopped, e.g. after
142	/// [TipWatcher::stop] or when its backend connection broke.
143	pub async fn changed(&mut self) -> anyhow::Result<BlockRef> {
144		self.rx.changed().await?;
145		// borrow_and_update marks the returned value as seen, so the next
146		// call only wakes for a tip newer than the one returned here.
147		Ok(*self.rx.borrow_and_update())
148	}
149}
bark/daemon/tip_watcher/mod.rs

bark/daemon/tip_watcher/
mod.rs
