urma_node/
source.rs

1use crate::{
2    Error,
3    model::{ChainPoint, CommitSourceStatus},
4    recorded::RecordedError,
5};
6use bitcoin::{BlockHash, Txid, block::Header};
7use std::{
8    net::SocketAddr,
9    path::{Path, PathBuf},
10    sync::{
11        Arc, Mutex, MutexGuard,
12        atomic::{AtomicBool, Ordering},
13    },
14    time::{Duration, Instant},
15};
16use urma_chain::{decode, observation::Chain};
17use urma_runtime::{
18    node::{Node, NodeConfig},
19    p2p::{Anchor, HeaderBootstrap, P2pProvider},
20};
21
22/// Port implemented by a chain verifier, not by an untrusted remote JSON API.
23/// The implementation owns header difficulty, PoW and branch selection. Bodies
24/// remain untrusted bytes and are checked again against the selected header.
25/// Provider-claimed observations used by autonomous clients are a different
26/// contract and do not enter this port by relabelling them as verified.
27pub trait ChainSource: Send + Sync {
28    fn genesis(&self) -> Result<BlockHash, Error>;
29    fn tip(&self) -> Result<ChainPoint, Error>;
30    fn header(&self, height: u64) -> Result<Header, Error>;
31    fn block(&self, height: u64) -> Result<Vec<u8>, Error>;
32    /// Connected chain peers now; 0 for sources without a peer network.
33    fn chain_peers(&self) -> Result<u64, Error> {
34        Ok(0)
35    }
36}
37
38/// Untrusted auxiliary parent bytes, not a chain verifier. The chain getter
39/// reads local configuration without I/O. Each fetch must bound its own I/O;
40/// Engine cancellation is cooperative and cannot interrupt a running fetch.
41pub trait CommitSource {
42    fn chain(&self) -> Chain;
43    fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error>;
44    /// A liveness check for status, rate-limited by the source itself. `None`
45    /// when no fresh check ran (or the source has none); otherwise its result.
46    fn probe(&self) -> Option<Result<(), Error>> {
47        None
48    }
49}
50
51pub(crate) fn check_cancelled(cancelled: &AtomicBool) -> Result<(), Error> {
52    if cancelled.load(Ordering::Acquire) {
53        return Err(Error::Busy("dependency acquisition cancelled".into()));
54    }
55    Ok(())
56}
57
58/// Decode with the Engine's chain decoder, so Litecoin MWEB/HogEx encodings
59/// are accepted, and require the computed TXID to equal the requested one.
60pub fn verify_parent(chain: Chain, txid: Txid, raw: &[u8]) -> Result<(), Error> {
61    let transaction = decode::transaction(raw, chain)
62        .map_err(|error| Error::Invalid(format!("commit parent bytes do not decode: {error}")))?;
63    if transaction.compute_txid() != txid {
64        return Err(Error::Invalid(
65            "commit parent bytes do not match the requested TXID".into(),
66        ));
67    }
68    Ok(())
69}
70
71/// Share an explicitly configured, owner-provided runtime Node. Constructing
72/// this wrapper or reading its chain creates no Node, connection or defaults.
73/// Only fetch calls the existing Node; no header/placement authority is added.
74pub struct RuntimeCommitSource {
75    pub node: Arc<Node>,
76}
77
78impl CommitSource for RuntimeCommitSource {
79    fn chain(&self) -> Chain {
80        self.node.chain()
81    }
82
83    /// Preserve the original returned hex bytes, rather than reserializing a
84    /// normalized transaction. The 16M-character/8M-byte admission cap applies
85    /// after Node::call: its transport may allocate/decode first and owns its
86    /// timeouts. There is no global deadline or transport-allocation promise.
87    /// Absence is the runtime's explicit Missing result or a local JSON-RPC
88    /// error with code -5 (no such mempool/blockchain transaction, including
89    /// lookups a node without -txindex cannot answer), per is_missing_transaction.
90    /// Other runtime errors become CommitSource without text matching.
91    fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error> {
92        check_cancelled(cancelled)?;
93        fetch_from_node(&self.node, txid)
94    }
95}
96
97/// Optional local Litecoin Core: it may start, stop or restart at any time.
98/// Holds only the URL and cookie path; every lookup and probe reads the cookie,
99/// and a node is rebuilt and its txindex verified whenever the cookie changes.
100pub struct LocalRpcCommitSource {
101    chain: Chain,
102    rpc_url: String,
103    cookie_file: PathBuf,
104    probe_interval: Duration,
105    verified: Mutex<Option<(Vec<u8>, Arc<Node>)>>,
106    probed: Mutex<Option<Instant>>,
107}
108
109impl LocalRpcCommitSource {
110    pub const PROBE_INTERVAL: Duration = Duration::from_secs(10);
111
112    pub fn new(chain: Chain, rpc_url: String, cookie_file: PathBuf) -> Self {
113        Self {
114            chain,
115            rpc_url,
116            cookie_file,
117            probe_interval: Self::PROBE_INTERVAL,
118            verified: Mutex::new(None),
119            probed: Mutex::new(None),
120        }
121    }
122
123    pub fn with_probe_interval(mut self, interval: Duration) -> Self {
124        self.probe_interval = interval;
125        self
126    }
127
128    fn node(&self) -> Result<Arc<Node>, Error> {
129        let cookie = std::fs::read(&self.cookie_file).map_err(|_| {
130            Error::CommitSource("cookie file missing: litecoind not running?".into())
131        })?;
132        let mut verified = self
133            .verified
134            .lock()
135            .map_err(|_| Error::Invalid("local RPC commit source poisoned".into()))?;
136        if let Some((seen, node)) = verified.as_ref() {
137            if *seen == cookie {
138                return Ok(Arc::clone(node));
139            }
140        }
141        *verified = None;
142        let node = Node::connect(NodeConfig {
143            chain: self.chain,
144            rpc_url: self.rpc_url.clone(),
145            cookie_file: self.cookie_file.clone(),
146        })
147        .and_then(|node| node.require_txindex().map(|()| Arc::new(node)))
148        .map_err(|error| Error::CommitSource(error.to_string()))?;
149        *verified = Some((cookie, Arc::clone(&node)));
150        Ok(node)
151    }
152
153    fn probe_due(&self) -> Result<bool, Error> {
154        let mut probed = self
155            .probed
156            .lock()
157            .map_err(|_| Error::Invalid("local RPC probe clock poisoned".into()))?;
158        let due = probed.is_none_or(|last| last.elapsed() >= self.probe_interval);
159        if due {
160            *probed = Some(Instant::now());
161        }
162        Ok(due)
163    }
164}
165
166impl CommitSource for LocalRpcCommitSource {
167    fn chain(&self) -> Chain {
168        self.chain
169    }
170
171    fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error> {
172        check_cancelled(cancelled)?;
173        let node = self.node()?;
174        fetch_from_node(&node, txid)
175    }
176
177    fn probe(&self) -> Option<Result<(), Error>> {
178        match self.probe_due() {
179            Ok(false) => None,
180            Ok(true) => Some(self.node().and_then(|node| {
181                node.call("getblockcount", &[])
182                    .map(|_| ())
183                    .map_err(|error| Error::CommitSource(error.to_string()))
184            })),
185            Err(error) => Some(Err(error)),
186        }
187    }
188}
189
190fn fetch_from_node(node: &Node, txid: Txid) -> Result<Option<Vec<u8>>, Error> {
191    let result = match node.call(
192        "getrawtransaction",
193        &[serde_json::json!(txid), serde_json::json!(false)],
194    ) {
195        Ok(result) => result,
196        Err(error) if error.is_missing_transaction() => return Ok(None),
197        Err(error) => return Err(Error::CommitSource(error.to_string())),
198    };
199    let encoded = result
200        .as_str()
201        .ok_or_else(|| Error::Invalid("commit source must return transaction hex".into()))?;
202    if encoded.len() > 16_000_000 || encoded.len() % 2 != 0 {
203        return Err(Error::Invalid(
204            "commit source hex exceeds capacity or has odd length".into(),
205        ));
206    }
207    let raw = hex::decode(encoded)
208        .map_err(|error| Error::Invalid(format!("commit source hex: {error}")))?;
209    Ok(Some(raw))
210}
211
212/// Default public providers for one chain, without a light client. Creating
213/// it does no I/O; the first fetch builds the Node. A failed build is not
214/// rebuilt for RETRY and its error is returned meanwhile. Untrusted bytes.
215pub struct PublicCommitSource {
216    chain: Chain,
217    cache_dir: PathBuf,
218    state: Mutex<PublicState>,
219}
220
221struct PublicState {
222    node: Option<Arc<Node>>,
223    failure: Option<(Instant, String)>,
224}
225
226impl PublicCommitSource {
227    const RETRY: Duration = Duration::from_secs(60);
228
229    pub fn new(chain: Chain, cache_dir: PathBuf) -> Self {
230        Self {
231            chain,
232            cache_dir,
233            state: Mutex::new(PublicState {
234                node: None,
235                failure: None,
236            }),
237        }
238    }
239
240    fn node(&self) -> Result<Arc<Node>, Error> {
241        let mut state = self
242            .state
243            .lock()
244            .map_err(|_| Error::Invalid("public commit source poisoned".into()))?;
245        if let Some(node) = state.node.as_ref() {
246            return Ok(Arc::clone(node));
247        }
248        if let Some((failed, message)) = state.failure.as_ref() {
249            if failed.elapsed() < Self::RETRY {
250                return Err(Error::CommitSource(message.clone()));
251            }
252        }
253        match Node::public_pinned(self.chain, &self.cache_dir) {
254            Ok(node) => {
255                let node = Arc::new(node);
256                state.node = Some(Arc::clone(&node));
257                state.failure = None;
258                Ok(node)
259            }
260            Err(error) => {
261                let message = error.to_string();
262                state.failure = Some((Instant::now(), message.clone()));
263                Err(Error::CommitSource(message))
264            }
265        }
266    }
267}
268
269impl CommitSource for PublicCommitSource {
270    fn chain(&self) -> Chain {
271        self.chain
272    }
273
274    fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error> {
275        check_cancelled(cancelled)?;
276        let node = self.node()?;
277        fetch_from_node(&node, txid)
278    }
279}
280
281/// Ordered labelled sources tried until one returns bytes. A miss or error
282/// moves to the next source; the set errors only when every source errored.
283/// The error table is locked only to read or write, never across a fetch.
284pub struct CommitSources {
285    chain: Chain,
286    entries: Vec<(String, Box<dyn CommitSource + Send + Sync>)>,
287    errors: Mutex<Vec<Option<RecordedError>>>,
288}
289
290impl CommitSources {
291    pub fn new(chain: Chain) -> Self {
292        Self {
293            chain,
294            entries: Vec::new(),
295            errors: Mutex::new(Vec::new()),
296        }
297    }
298
299    pub fn push(&mut self, label: String, source: Box<dyn CommitSource + Send + Sync>) {
300        self.entries.push((label, source));
301    }
302
303    pub fn is_empty(&self) -> bool {
304        self.entries.is_empty()
305    }
306
307    pub fn status(&self) -> Result<Vec<CommitSourceStatus>, Error> {
308        for (index, entry) in self.entries.iter().enumerate() {
309            if let Some(probed) = entry.1.probe() {
310                self.record(index, probed.err().map(|error| RecordedError::now(&error)))?;
311            }
312        }
313        let errors = self.errors()?;
314        Ok(self
315            .entries
316            .iter()
317            .zip(errors.iter())
318            .map(|(entry, error)| {
319                let (last_error, last_error_at) = RecordedError::split(error.clone());
320                CommitSourceStatus {
321                    label: entry.0.clone(),
322                    last_error,
323                    last_error_at,
324                }
325            })
326            .collect())
327    }
328
329    pub(crate) fn source_chains(&self) -> Vec<Chain> {
330        self.entries.iter().map(|entry| entry.1.chain()).collect()
331    }
332
333    fn errors(&self) -> Result<MutexGuard<'_, Vec<Option<RecordedError>>>, Error> {
334        let mut errors = self
335            .errors
336            .lock()
337            .map_err(|_| Error::Invalid("commit source status poisoned".into()))?;
338        errors.resize(self.entries.len(), None);
339        Ok(errors)
340    }
341
342    fn verified(&self, txid: Txid, raw: Option<Vec<u8>>) -> Result<Option<Vec<u8>>, Error> {
343        let Some(raw) = raw else {
344            return Ok(None);
345        };
346        verify_parent(self.chain, txid, &raw)?;
347        Ok(Some(raw))
348    }
349
350    fn record(&self, index: usize, error: Option<RecordedError>) -> Result<(), Error> {
351        let mut errors = self.errors()?;
352        let slot = errors
353            .get_mut(index)
354            .ok_or_else(|| Error::Invalid("commit source status out of range".into()))?;
355        *slot = error;
356        Ok(())
357    }
358}
359
360impl CommitSource for CommitSources {
361    fn chain(&self) -> Chain {
362        self.chain
363    }
364
365    fn fetch(&self, txid: Txid, cancelled: &AtomicBool) -> Result<Option<Vec<u8>>, Error> {
366        let mut last = None;
367        let mut failures = 0;
368        for (index, entry) in self.entries.iter().enumerate() {
369            check_cancelled(cancelled)?;
370            let fetched = entry
371                .1
372                .fetch(txid, cancelled)
373                .and_then(|raw| self.verified(txid, raw));
374            check_cancelled(cancelled)?;
375            match fetched {
376                Ok(Some(raw)) => {
377                    self.record(index, None)?;
378                    return Ok(Some(raw));
379                }
380                Ok(None) => self.record(index, None)?,
381                Err(error) => {
382                    self.record(index, Some(RecordedError::now(&error)))?;
383                    failures += 1;
384                    last = Some(error);
385                }
386            }
387        }
388        let Some(error) = last else {
389            return Ok(None);
390        };
391        if failures == self.entries.len() {
392            return Err(error);
393        }
394        Ok(None)
395    }
396}
397
398/// Checkpoint-relative Litecoin light client. Does not validate scripts, UTXOs
399/// or MWEB consensus. Opening it does not start network connections; the first
400/// sync attempt does. An application start before the configured anchor is an
401/// explicit error, never silently replaced with the anchor's height.
402pub struct LitecoinSource {
403    chain: Chain,
404    provider: P2pProvider,
405}
406
407impl LitecoinSource {
408    pub fn open(
409        chain: Chain,
410        cache: &Path,
411        anchor: Anchor,
412        peers: Vec<SocketAddr>,
413    ) -> Result<Self, Error> {
414        if !matches!(chain, Chain::LitecoinTestnet | Chain::LitecoinMainnet) {
415            return Err(Error::Invalid(
416                "Litecoin source requires a Litecoin network".into(),
417            ));
418        }
419        Ok(Self {
420            chain,
421            provider: P2pProvider::with_peers(chain, cache, anchor, peers).map_err(source_error)?,
422        })
423    }
424
425    /// Open a Litecoin source with authenticated preceding header context.
426    /// The operator trusts the bootstrap anchor's chain membership and height;
427    /// the context does not prove pre-anchor consensus or earlier scan coverage.
428    /// This source does not validate scripts, UTXOs or MWEB consensus. Opening
429    /// only loads the local cache; the first sync attempt starts connections.
430    pub fn open_bootstrap(
431        cache: &Path,
432        bootstrap: HeaderBootstrap,
433        peers: Vec<SocketAddr>,
434    ) -> Result<Self, Error> {
435        let chain = bootstrap.chain();
436        if !matches!(chain, Chain::LitecoinTestnet | Chain::LitecoinMainnet) {
437            return Err(Error::Invalid(
438                "Litecoin source requires a Litecoin network".into(),
439            ));
440        }
441        Ok(Self {
442            chain,
443            provider: P2pProvider::with_bootstrap(cache, bootstrap, peers).map_err(source_error)?,
444        })
445    }
446}
447
448fn source_error(e: urma_runtime::error::Error) -> Error {
449    Error::Source(Box::new(e))
450}
451
452impl ChainSource for LitecoinSource {
453    fn genesis(&self) -> Result<BlockHash, Error> {
454        Ok(self
455            .chain
456            .genesis()
457            .map_err(|e| Error::Invalid(e.to_string()))?
458            .0)
459    }
460    fn tip(&self) -> Result<ChainPoint, Error> {
461        let (height, header) = self.provider.verified_tip().map_err(source_error)?;
462        Ok(ChainPoint {
463            height,
464            hash: header.block_hash(),
465        })
466    }
467    fn header(&self, height: u64) -> Result<Header, Error> {
468        self.provider.verified_header(height).map_err(source_error)
469    }
470    fn block(&self, height: u64) -> Result<Vec<u8>, Error> {
471        Ok(self
472            .provider
473            .verified_block(height)
474            .map_err(source_error)?
475            .raw()
476            .to_vec())
477    }
478    fn chain_peers(&self) -> Result<u64, Error> {
479        u64::try_from(self.provider.progress().peers_connected)
480            .map_err(|error| Error::Invalid(format!("chain peer count: {error}")))
481    }
482}
urma_node/source.rs

urma_node/
source.rs
