urma_runtime/p2p/
provider.rs

1use crate::config::{P2P_BLOCK_TIMEOUT_SECS, P2P_TIP_REFRESH_SECS};
2use crate::error::{Context, Error, ensure};
3use crate::light::{LightSync, Progress};
4use crate::p2p::HeaderBootstrap;
5use crate::p2p::blocks::BlockCache;
6use crate::p2p::headers::{Anchor, HeaderChain};
7use crate::p2p::source::{VerifiedBlock, decode_verified};
8use crate::p2p::worker::{self, Command, Freshness, Shared, Status, locked};
9use crate::transport::{BlockEncoding, Evidence, Provider};
10use bitcoin::BlockHash;
11use bitcoin::block::Header;
12use serde_json::{Value, json};
13use std::net::SocketAddr;
14use std::path::Path;
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::mpsc::{self, SyncSender, TrySendError};
17use std::sync::{Arc, Mutex};
18use std::time::{Duration, Instant};
19use urma_chain::observation::Chain;
20
21enum Engine {
22    Cold,
23    Running(SyncSender<Command>),
24}
25
26/// P2P header view rooted in a trusted checkpoint and existing local cache.
27/// Typed results establish the limited verification described by `VerifiedBlock`.
28pub struct P2pProvider {
29    chain: Chain,
30    shared: Arc<Shared>,
31    engine: Mutex<Engine>,
32}
33
34impl P2pProvider {
35    pub fn new(chain: Chain, cache_dir: &Path) -> Result<Self, Error> {
36        Self::with_peers(chain, cache_dir, Anchor::embedded(chain)?, Vec::new())
37    }
38
39    pub fn with_peers(
40        chain: Chain,
41        cache_dir: &Path,
42        anchor: Anchor,
43        pinned: Vec<SocketAddr>,
44    ) -> Result<Self, Error> {
45        let started = Instant::now();
46        let loaded = HeaderChain::open_at(chain, cache_dir, anchor);
47        tracing::debug!(
48            target: "urma_startup",
49            chain = chain.label(),
50            stage = "header_chain_open",
51            elapsed_ms = started.elapsed().as_secs_f64() * 1000.0,
52            result = ?loaded.as_ref().map(HeaderChain::synced_count),
53            "light client stage completed"
54        );
55        let headers = loaded?;
56        Ok(Self::from_headers(chain, headers, pinned, cache_dir))
57    }
58
59    /// Open a checkpoint-relative view with authenticated preceding context.
60    /// The operator still trusts the bootstrap anchor's chain membership and
61    /// height; its context does not establish full consensus or earlier scan
62    /// coverage. Construction opens the local cache without starting a worker
63    /// or network connections. `warm` or the first verified read starts it.
64    pub fn with_bootstrap(
65        cache_dir: &Path,
66        bootstrap: HeaderBootstrap,
67        pinned: Vec<SocketAddr>,
68    ) -> Result<Self, Error> {
69        let chain = bootstrap.chain();
70        let headers = HeaderChain::open_bootstrap(cache_dir, bootstrap)?;
71        Ok(Self::from_headers(chain, headers, pinned, cache_dir))
72    }
73
74    fn from_headers(
75        chain: Chain,
76        headers: HeaderChain,
77        pinned: Vec<SocketAddr>,
78        cache_dir: &Path,
79    ) -> Self {
80        let shared = Arc::new(Shared {
81            chain,
82            pinned,
83            cache_dir: cache_dir.to_owned(),
84            headers: Mutex::new(headers),
85            blocks: Mutex::new(BlockCache::new()),
86            status: Mutex::new(Status {
87                peers: 0,
88                target: 0,
89                synced: false,
90                refreshed: Freshness::Never,
91            }),
92            stop: AtomicBool::new(false),
93        });
94        Self {
95            chain,
96            shared,
97            engine: Mutex::new(Engine::Cold),
98        }
99    }
100
101    pub fn warm(&self) -> Result<(), Error> {
102        let mut engine = locked(&self.engine, "engine");
103        match &*engine {
104            Engine::Running(_commands) => return Ok(()),
105            Engine::Cold => (),
106        }
107        let (sender, receiver) = mpsc::sync_channel(16);
108        let background = self.shared.clone();
109        std::thread::Builder::new()
110            .name("urma-p2p".into())
111            .spawn(move || worker::run(background, receiver))?;
112        *engine = Engine::Running(sender);
113        tracing::debug!(chain = self.chain.label(), "light client worker started");
114        Ok(())
115    }
116
117    fn ignite(&self) {
118        match self.warm() {
119            Ok(()) => (),
120            Err(error) => tracing::warn!(%error, "light client worker could not start"),
121        }
122    }
123
124    pub fn progress(&self) -> Progress {
125        let headers = locked(&self.shared.headers, "headers");
126        let status = locked(&self.shared.status, "status");
127        Progress {
128            headers_synced: headers.synced_count(),
129            headers_target: status.target.max(headers.tip_height()) - headers.start_height(),
130            peers_connected: status.peers,
131            synced: status.synced,
132        }
133    }
134
135    /// Absolute height header sync heads for: the best height connected peers
136    /// announced, never below the local tip. Announced, not proven. `None`
137    /// while no peer is connected or none has announced a height.
138    pub fn target_height(&self) -> Option<u64> {
139        let headers = locked(&self.shared.headers, "headers");
140        let status = locked(&self.shared.status, "status");
141        (status.peers > 0 && status.target > 0).then(|| status.target.max(headers.tip_height()))
142    }
143
144    pub fn state(&self) -> LightSync {
145        let progress = self.progress();
146        match &*locked(&self.engine, "engine") {
147            Engine::Cold => LightSync::Cold(progress),
148            Engine::Running(_commands) => LightSync::Running(progress),
149        }
150    }
151
152    pub fn synced(&self) -> bool {
153        locked(&self.shared.status, "status").synced
154    }
155
156    pub fn ready(&self) -> bool {
157        let status = locked(&self.shared.status, "status");
158        status.synced && status.peers > 0
159    }
160
161    fn command(&self, command: Command) -> Result<(), Error> {
162        let sender = match &*locked(&self.engine, "engine") {
163            Engine::Cold => {
164                return Err(Error::Unsupported(
165                    "light client worker has not been started".into(),
166                ));
167            }
168            Engine::Running(sender) => sender.clone(),
169        };
170        match sender.try_send(command) {
171            Ok(()) => Ok(()),
172            Err(TrySendError::Full(_command)) => {
173                Err(Error::Capacity("light client command queue is full".into()))
174            }
175            Err(TrySendError::Disconnected(_command)) => {
176                Err(Error::Unsupported("light client worker stopped".into()))
177            }
178        }
179    }
180
181    fn source_ready(&self) -> Result<(), Error> {
182        self.warm()?;
183        if !self.ready() {
184            return Err(Error::Unsupported(
185                "verified P2P source is not ready; retry after header sync and peer connection"
186                    .into(),
187            ));
188        }
189        Ok(())
190    }
191
192    /// Snapshot of the active header tip, relative to the configured trusted
193    /// checkpoint and local header cache, not full consensus validation.
194    /// Starts the worker if cold and fails immediately while it is not ready.
195    /// Stale-tip refresh waits at most `P2P_BLOCK_TIMEOUT_SECS` for its reply;
196    /// a full command queue is refused immediately. No other source is used.
197    pub fn verified_tip(&self) -> Result<(u64, Header), Error> {
198        self.source_ready()?;
199        self.refresh_if_stale()?;
200        self.source_ready()?;
201        let headers = locked(&self.shared.headers, "headers");
202        let height = headers.tip_height();
203        Ok((height, headers.header_at(height)?))
204    }
205
206    /// Active header at `height`, under the same trust and readiness contract
207    /// as `verified_tip()`. Heights before the checkpoint are unavailable.
208    pub fn verified_header(&self, height: u64) -> Result<Header, Error> {
209        self.source_ready()?;
210        self.refresh_if_stale()?;
211        self.source_ready()?;
212        locked(&self.shared.headers, "headers").header_at(height)
213    }
214
215    /// Fetch by the captured active header's hash, check block integrity, and
216    /// refuse the result if that header was replaced or removed during fetch.
217    /// Retains original bytes; see `VerifiedBlock` for verification limits.
218    /// Readiness and queue admission fail fast. Refresh and block replies each
219    /// wait at most `P2P_BLOCK_TIMEOUT_SECS`; no router or fallback is involved.
220    pub fn verified_block(&self, height: u64) -> Result<VerifiedBlock, Error> {
221        let header = self.verified_header(height)?;
222        let raw = self.fetch_raw(header.block_hash())?;
223        let block = decode_verified(self.chain, height, header, raw)?;
224        let headers = locked(&self.shared.headers, "headers");
225        ensure!(
226            headers.header_at(height)? == header,
227            "active verified header changed during block fetch at height {height}"
228        );
229        Ok(block)
230    }
231
232    fn fetch_raw(&self, hash: BlockHash) -> Result<Vec<u8>, Error> {
233        let (reply, answer) = mpsc::sync_channel(1);
234        self.command(Command::Block { hash, reply })?;
235        answer
236            .recv_timeout(Duration::from_secs(P2P_BLOCK_TIMEOUT_SECS))
237            .map_err(|error| Error::Unsupported(format!("block fetch did not complete: {error}")))?
238    }
239
240    fn refresh_if_stale(&self) -> Result<(), Error> {
241        let stale = match locked(&self.shared.status, "status").refreshed {
242            Freshness::Never => true,
243            Freshness::At(when) => when.elapsed() > Duration::from_secs(P2P_TIP_REFRESH_SECS),
244        };
245        if !stale {
246            return Ok(());
247        }
248        let (reply, answer) = mpsc::sync_channel(1);
249        self.command(Command::Refresh { reply })?;
250        answer
251            .recv_timeout(Duration::from_secs(P2P_BLOCK_TIMEOUT_SECS))
252            .map_err(|error| {
253                Error::Unsupported(format!("header refresh did not complete: {error}"))
254            })?
255    }
256
257    fn chain_info(&self) -> Result<Value, Error> {
258        self.refresh_if_stale()?;
259        let headers = locked(&self.shared.headers, "headers");
260        let name = match self.chain {
261            Chain::BitcoinMainnet | Chain::LitecoinMainnet => "main",
262            Chain::LitecoinTestnet | Chain::BitcoinTestnet4 => "test",
263            Chain::BitcoinRegtest => "regtest",
264        };
265        Ok(json!({
266            "chain": name,
267            "blocks": headers.tip_height(),
268            "headers": headers.tip_height(),
269            "bestblockhash": headers.tip_hash().to_string(),
270            "initialblockdownload": false,
271        }))
272    }
273
274    fn block_hash(&self, args: &[Value]) -> Result<Value, Error> {
275        let height = args
276            .first()
277            .context("missing block height")?
278            .as_u64()
279            .context("block height must be a non-negative integer")?;
280        self.refresh_if_stale()?;
281        let headers = locked(&self.shared.headers, "headers");
282        Ok(json!(headers.hash_at(height)?.to_string()))
283    }
284
285    fn block_header(&self, args: &[Value]) -> Result<Value, Error> {
286        let hash = requested_hash(args)?;
287        let headers = locked(&self.shared.headers, "headers");
288        let height = headers.height_of(hash)?;
289        let header = headers.header_at(height)?;
290        Ok(json!({
291            "hash": hash.to_string(),
292            "confirmations": headers.tip_height() - height + 1,
293            "height": height,
294            "version": header.version.to_consensus(),
295            "merkleroot": header.merkle_root.to_string(),
296            "time": header.time,
297            "nonce": header.nonce,
298            "bits": format!("{:08x}", header.bits.to_consensus()),
299            "previousblockhash": header.prev_blockhash.to_string(),
300        }))
301    }
302
303    fn block(&self, args: &[Value]) -> Result<Value, Error> {
304        let hash = requested_hash(args)?;
305        ensure!(
306            args.get(1).context("missing block verbosity")? == &json!(0),
307            "the light client serves raw blocks only (verbosity 0)"
308        );
309        let raw = self.fetch_raw(hash)?;
310        Ok(json!(hex::encode(raw)))
311    }
312}
313
314fn requested_hash(args: &[Value]) -> Result<BlockHash, Error> {
315    Ok(args
316        .first()
317        .context("missing block hash")?
318        .as_str()
319        .context("block hash must be a string")?
320        .parse()?)
321}
322
323impl Provider for P2pProvider {
324    fn label(&self) -> String {
325        format!("p2p light client ({})", self.chain.label())
326    }
327
328    fn evidence(&self) -> Evidence {
329        Evidence::LightClientInclusion
330    }
331
332    fn block_encoding(&self) -> BlockEncoding {
333        BlockEncoding::Esplora
334    }
335
336    fn supports(&self, method: &str) -> bool {
337        match method {
338            "getblock" | "getblockheader" => {
339                self.ignite();
340                self.ready()
341            }
342            "getblockchaininfo" | "getblockhash" => self.ready(),
343            _other => false,
344        }
345    }
346
347    fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
348        ensure!(
349            chain.genesis()? == self.chain.genesis()?,
350            "light client serves {} only",
351            self.chain.label()
352        );
353        match method {
354            "getblockchaininfo" => self.chain_info(),
355            "getblockhash" => self.block_hash(args),
356            "getblockheader" => self.block_header(args),
357            "getblock" => self.block(args),
358            other => Err(Error::Unsupported(format!(
359                "the peer-to-peer light client does not serve {other}"
360            ))),
361        }
362    }
363}
364
365impl Provider for Arc<P2pProvider> {
366    fn label(&self) -> String {
367        P2pProvider::label(self)
368    }
369
370    fn evidence(&self) -> Evidence {
371        P2pProvider::evidence(self)
372    }
373
374    fn block_encoding(&self) -> BlockEncoding {
375        P2pProvider::block_encoding(self)
376    }
377
378    fn supports(&self, method: &str) -> bool {
379        P2pProvider::supports(self, method)
380    }
381
382    fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
383        P2pProvider::call(self, chain, method, args)
384    }
385}
386
387impl Drop for P2pProvider {
388    fn drop(&mut self) {
389        self.shared.stop.store(true, Ordering::Release);
390    }
391}
392
393#[cfg(test)]
394#[path = "source_tests.rs"]
395mod source_tests;
urma_runtime/p2p/provider.rs

urma_runtime/p2p/
provider.rs
