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    pub fn state(&self) -> LightSync {
136        let progress = self.progress();
137        match &*locked(&self.engine, "engine") {
138            Engine::Cold => LightSync::Cold(progress),
139            Engine::Running(_commands) => LightSync::Running(progress),
140        }
141    }
142
143    pub fn synced(&self) -> bool {
144        locked(&self.shared.status, "status").synced
145    }
146
147    pub fn ready(&self) -> bool {
148        let status = locked(&self.shared.status, "status");
149        status.synced && status.peers > 0
150    }
151
152    fn command(&self, command: Command) -> Result<(), Error> {
153        let sender = match &*locked(&self.engine, "engine") {
154            Engine::Cold => {
155                return Err(Error::Unsupported(
156                    "light client worker has not been started".into(),
157                ));
158            }
159            Engine::Running(sender) => sender.clone(),
160        };
161        match sender.try_send(command) {
162            Ok(()) => Ok(()),
163            Err(TrySendError::Full(_command)) => {
164                Err(Error::Capacity("light client command queue is full".into()))
165            }
166            Err(TrySendError::Disconnected(_command)) => {
167                Err(Error::Unsupported("light client worker stopped".into()))
168            }
169        }
170    }
171
172    fn source_ready(&self) -> Result<(), Error> {
173        self.warm()?;
174        if !self.ready() {
175            return Err(Error::Unsupported(
176                "verified P2P source is not ready; retry after header sync and peer connection"
177                    .into(),
178            ));
179        }
180        Ok(())
181    }
182
183    /// Snapshot of the active header tip, relative to the configured trusted
184    /// checkpoint and local header cache, not full consensus validation.
185    /// Starts the worker if cold and fails immediately while it is not ready.
186    /// Stale-tip refresh waits at most `P2P_BLOCK_TIMEOUT_SECS` for its reply;
187    /// a full command queue is refused immediately. No other source is used.
188    pub fn verified_tip(&self) -> Result<(u64, Header), Error> {
189        self.source_ready()?;
190        self.refresh_if_stale()?;
191        self.source_ready()?;
192        let headers = locked(&self.shared.headers, "headers");
193        let height = headers.tip_height();
194        Ok((height, headers.header_at(height)?))
195    }
196
197    /// Active header at `height`, under the same trust and readiness contract
198    /// as `verified_tip()`. Heights before the checkpoint are unavailable.
199    pub fn verified_header(&self, height: u64) -> Result<Header, Error> {
200        self.source_ready()?;
201        self.refresh_if_stale()?;
202        self.source_ready()?;
203        locked(&self.shared.headers, "headers").header_at(height)
204    }
205
206    /// Fetch by the captured active header's hash, check block integrity, and
207    /// refuse the result if that header was replaced or removed during fetch.
208    /// Retains original bytes; see `VerifiedBlock` for verification limits.
209    /// Readiness and queue admission fail fast. Refresh and block replies each
210    /// wait at most `P2P_BLOCK_TIMEOUT_SECS`; no router or fallback is involved.
211    pub fn verified_block(&self, height: u64) -> Result<VerifiedBlock, Error> {
212        let header = self.verified_header(height)?;
213        let raw = self.fetch_raw(header.block_hash())?;
214        let block = decode_verified(self.chain, height, header, raw)?;
215        let headers = locked(&self.shared.headers, "headers");
216        ensure!(
217            headers.header_at(height)? == header,
218            "active verified header changed during block fetch at height {height}"
219        );
220        Ok(block)
221    }
222
223    fn fetch_raw(&self, hash: BlockHash) -> Result<Vec<u8>, Error> {
224        let (reply, answer) = mpsc::sync_channel(1);
225        self.command(Command::Block { hash, reply })?;
226        answer
227            .recv_timeout(Duration::from_secs(P2P_BLOCK_TIMEOUT_SECS))
228            .map_err(|error| Error::Unsupported(format!("block fetch did not complete: {error}")))?
229    }
230
231    fn refresh_if_stale(&self) -> Result<(), Error> {
232        let stale = match locked(&self.shared.status, "status").refreshed {
233            Freshness::Never => true,
234            Freshness::At(when) => when.elapsed() > Duration::from_secs(P2P_TIP_REFRESH_SECS),
235        };
236        if !stale {
237            return Ok(());
238        }
239        let (reply, answer) = mpsc::sync_channel(1);
240        self.command(Command::Refresh { reply })?;
241        answer
242            .recv_timeout(Duration::from_secs(P2P_BLOCK_TIMEOUT_SECS))
243            .map_err(|error| {
244                Error::Unsupported(format!("header refresh did not complete: {error}"))
245            })?
246    }
247
248    fn chain_info(&self) -> Result<Value, Error> {
249        self.refresh_if_stale()?;
250        let headers = locked(&self.shared.headers, "headers");
251        let name = match self.chain {
252            Chain::BitcoinMainnet | Chain::LitecoinMainnet => "main",
253            Chain::LitecoinTestnet | Chain::BitcoinTestnet4 => "test",
254            Chain::BitcoinRegtest => "regtest",
255        };
256        Ok(json!({
257            "chain": name,
258            "blocks": headers.tip_height(),
259            "headers": headers.tip_height(),
260            "bestblockhash": headers.tip_hash().to_string(),
261            "initialblockdownload": false,
262        }))
263    }
264
265    fn block_hash(&self, args: &[Value]) -> Result<Value, Error> {
266        let height = args
267            .first()
268            .context("missing block height")?
269            .as_u64()
270            .context("block height must be a non-negative integer")?;
271        self.refresh_if_stale()?;
272        let headers = locked(&self.shared.headers, "headers");
273        Ok(json!(headers.hash_at(height)?.to_string()))
274    }
275
276    fn block_header(&self, args: &[Value]) -> Result<Value, Error> {
277        let hash = requested_hash(args)?;
278        let headers = locked(&self.shared.headers, "headers");
279        let height = headers.height_of(hash)?;
280        let header = headers.header_at(height)?;
281        Ok(json!({
282            "hash": hash.to_string(),
283            "confirmations": headers.tip_height() - height + 1,
284            "height": height,
285            "version": header.version.to_consensus(),
286            "merkleroot": header.merkle_root.to_string(),
287            "time": header.time,
288            "nonce": header.nonce,
289            "bits": format!("{:08x}", header.bits.to_consensus()),
290            "previousblockhash": header.prev_blockhash.to_string(),
291        }))
292    }
293
294    fn block(&self, args: &[Value]) -> Result<Value, Error> {
295        let hash = requested_hash(args)?;
296        ensure!(
297            args.get(1).context("missing block verbosity")? == &json!(0),
298            "the light client serves raw blocks only (verbosity 0)"
299        );
300        let raw = self.fetch_raw(hash)?;
301        Ok(json!(hex::encode(raw)))
302    }
303}
304
305fn requested_hash(args: &[Value]) -> Result<BlockHash, Error> {
306    Ok(args
307        .first()
308        .context("missing block hash")?
309        .as_str()
310        .context("block hash must be a string")?
311        .parse()?)
312}
313
314impl Provider for P2pProvider {
315    fn label(&self) -> String {
316        format!("p2p light client ({})", self.chain.label())
317    }
318
319    fn evidence(&self) -> Evidence {
320        Evidence::LightClientInclusion
321    }
322
323    fn block_encoding(&self) -> BlockEncoding {
324        BlockEncoding::Esplora
325    }
326
327    fn supports(&self, method: &str) -> bool {
328        match method {
329            "getblock" | "getblockheader" => {
330                self.ignite();
331                self.ready()
332            }
333            "getblockchaininfo" | "getblockhash" => self.ready(),
334            _other => false,
335        }
336    }
337
338    fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
339        ensure!(
340            chain.genesis()? == self.chain.genesis()?,
341            "light client serves {} only",
342            self.chain.label()
343        );
344        match method {
345            "getblockchaininfo" => self.chain_info(),
346            "getblockhash" => self.block_hash(args),
347            "getblockheader" => self.block_header(args),
348            "getblock" => self.block(args),
349            other => Err(Error::Unsupported(format!(
350                "the peer-to-peer light client does not serve {other}"
351            ))),
352        }
353    }
354}
355
356impl Provider for Arc<P2pProvider> {
357    fn label(&self) -> String {
358        P2pProvider::label(self)
359    }
360
361    fn evidence(&self) -> Evidence {
362        P2pProvider::evidence(self)
363    }
364
365    fn block_encoding(&self) -> BlockEncoding {
366        P2pProvider::block_encoding(self)
367    }
368
369    fn supports(&self, method: &str) -> bool {
370        P2pProvider::supports(self, method)
371    }
372
373    fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
374        P2pProvider::call(self, chain, method, args)
375    }
376}
377
378impl Drop for P2pProvider {
379    fn drop(&mut self) {
380        self.shared.stop.store(true, Ordering::Release);
381    }
382}
383
384#[cfg(test)]
385#[path = "source_tests.rs"]
386mod source_tests;
urma_runtime/p2p/provider.rs

urma_runtime/p2p/
provider.rs
