1use crate::config;
2use crate::error::{Context, Error, ensure};
3use crate::light::LightSync;
4#[cfg(not(target_arch = "wasm32"))]
5use crate::pinning::Pins;
6use crate::transaction::decode;
7use crate::{
8 endpoints::{self, PublicEndpoint},
9 providers,
10 transport::{BlockEncoding, Evidence, Preference, Provider, Router, Standing},
11};
12use bitcoin::{Transaction, Txid};
13use bitcoincore_rpc::{Auth, Client, RpcApi};
14use serde::{Deserialize, Serialize};
15use serde_json::{Value, json};
16use std::{
17 collections::{HashMap, hash_map::Entry},
18 path::PathBuf,
19 sync::{Mutex, MutexGuard},
20 time::Instant,
21};
22use urma_chain::{
23 decode::{self as chain_decode, DecodeError},
24 observation::Chain,
25 validation::BlockValidationError,
26};
27
28#[derive(Clone)]
29pub struct NodeConfig {
30 pub chain: Chain,
31 pub rpc_url: String,
32 pub cookie_file: PathBuf,
33}
34
35pub struct Node {
36 chain: Chain,
37 backend: Backend,
38 light: providers::LightClient,
39 acknowledged: Mutex<HashMap<Txid, Acknowledged>>,
40}
41
42struct Acknowledged {
43 when: Instant,
44 provider: String,
45}
46
47enum Backend {
48 Local(Client),
49 Routed(Router),
50}
51
52pub struct Observed {
53 pub value: Value,
54 pub provider: String,
55 pub evidence: Evidence,
56}
57
58#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(tag = "status", rename_all = "snake_case")]
60pub enum Presence {
61 Missing,
62 Mempool,
63 Confirmed { height: u64, block_hash: String },
64}
65
66impl Node {
67 pub fn connect(config: NodeConfig) -> Result<Self, Error> {
68 let url = url::Url::parse(&config.rpc_url)?;
69 ensure!(
70 url.scheme() == "http"
71 && url.username().is_empty()
72 && url.password().iter().count() == 0
73 && url.query().iter().count() == 0
74 && url.fragment().iter().count() == 0
75 && url.path() == "/",
76 "RPC requires a local HTTP origin and cookie authentication"
77 );
78 let host = url
79 .host_str()
80 .context("missing RPC host")?
81 .trim_matches(['[', ']']);
82 ensure!(
83 host.parse::<std::net::IpAddr>()?.is_loopback(),
84 "RPC must use a loopback IP"
85 );
86 let client = Client::new(url.as_str(), Auth::CookieFile(config.cookie_file.clone()))?;
87 let node = Self {
88 chain: config.chain,
89 backend: Backend::Local(client),
90 light: providers::LightClient::Absent,
91 acknowledged: Mutex::new(HashMap::new()),
92 };
93 node.verify_network()?;
94 Ok(node)
95 }
96
97 pub fn public(chain: Chain) -> Result<Self, Error> {
98 Self::with_public_sources(chain, endpoints::defaults(chain))
99 }
100
101 #[cfg(not(target_arch = "wasm32"))]
102 pub fn public_in(chain: Chain, cache_dir: &std::path::Path) -> Result<Self, Error> {
103 Self::public_in_with_endpoints(chain, cache_dir, endpoints::defaults(chain))
104 }
105
106 #[cfg(not(target_arch = "wasm32"))]
107 pub fn public_in_with_endpoints(
108 chain: Chain,
109 cache_dir: &std::path::Path,
110 public: Vec<endpoints::PublicEndpoint>,
111 ) -> Result<Self, Error> {
112 Self::public_from(chain, cache_dir, public, Vec::new())
113 }
114
115 #[cfg(not(target_arch = "wasm32"))]
116 pub fn public_in_with_providers(
117 chain: Chain,
118 cache_dir: &std::path::Path,
119 additional: Vec<Box<dyn Provider>>,
120 ) -> Result<Self, Error> {
121 Self::public_from(chain, cache_dir, endpoints::defaults(chain), additional)
122 }
123
124 #[cfg(not(target_arch = "wasm32"))]
125 fn public_from(
126 chain: Chain,
127 cache_dir: &std::path::Path,
128 public: Vec<endpoints::PublicEndpoint>,
129 mut additional: Vec<Box<dyn Provider>>,
130 ) -> Result<Self, Error> {
131 let pins = std::sync::Arc::new(Pins::in_directory(cache_dir)?);
132 let light = providers::light_clients(chain, cache_dir)?;
133 additional.extend(providers::assemble(public, pins, &light)?);
134 let mut node = Self::with_providers(chain, additional)?;
135 node.light = light;
136 Ok(node)
137 }
138
139 #[cfg(not(target_arch = "wasm32"))]
140 pub fn public_pinned(chain: Chain, cache_dir: &std::path::Path) -> Result<Self, Error> {
141 let pins = std::sync::Arc::new(Pins::in_directory(cache_dir)?);
142 let public = providers::assemble(
143 endpoints::defaults(chain),
144 pins,
145 &providers::LightClient::Absent,
146 )?;
147 Self::with_providers(chain, public)
148 }
149
150 pub fn light_client_progress(&self) -> LightSync {
151 self.light.state()
152 }
153
154 pub fn warm_light_client(&self) -> Result<(), Error> {
155 self.light.warm()
156 }
157
158 pub fn with_public_sources(
159 chain: Chain,
160 endpoints: Vec<PublicEndpoint>,
161 ) -> Result<Self, Error> {
162 Self::with_providers(chain, providers::ephemeral(chain, endpoints)?)
163 }
164
165 pub fn with_providers(chain: Chain, providers: Vec<Box<dyn Provider>>) -> Result<Self, Error> {
166 let router = Router::new(providers)?;
167 tracing::debug!(providers = ?router.labels(), evidence = router.evidence().label(), "routed public node");
168 let node = Self {
169 chain,
170 backend: Backend::Routed(router),
171 light: providers::LightClient::Absent,
172 acknowledged: Mutex::new(HashMap::new()),
173 };
174 node.verify_network()?;
175 Ok(node)
176 }
177
178 pub fn is_public(&self) -> bool {
179 matches!(self.backend, Backend::Routed(_))
180 }
181
182 pub fn evidence(&self) -> Evidence {
183 match &self.backend {
184 Backend::Local(_) => Evidence::LocalValidatingNode,
185 Backend::Routed(router) => router.evidence(),
186 }
187 }
188
189 pub fn inclusion_evidence(&self) -> &'static str {
190 self.evidence().label()
191 }
192
193 pub fn provider_labels(&self) -> Vec<String> {
194 match &self.backend {
195 Backend::Local(_) => vec![config::LOCAL_PROVIDER_LABEL.to_owned()],
196 Backend::Routed(router) => router.labels(),
197 }
198 }
199
200 pub fn standings(&self) -> Vec<Standing> {
201 match &self.backend {
202 Backend::Local(_) => vec![Standing {
203 label: config::LOCAL_PROVIDER_LABEL.to_owned(),
204 evidence: Evidence::LocalValidatingNode,
205 failures: 0,
206 blocked_for: std::time::Duration::ZERO,
207 }],
208 Backend::Routed(router) => router.standings(),
209 }
210 }
211
212 pub fn observe(&self, method: &str, args: &[Value]) -> Result<Observed, Error> {
213 match &self.backend {
214 Backend::Local(client) => Ok(Observed {
215 value: client.call(method, args)?,
216 provider: config::LOCAL_PROVIDER_LABEL.to_owned(),
217 evidence: Evidence::LocalValidatingNode,
218 }),
219 Backend::Routed(router) => {
220 let answer = router.answer(self.chain, method, args)?;
221 Ok(Observed {
222 value: answer.value,
223 provider: answer.label,
224 evidence: answer.evidence,
225 })
226 }
227 }
228 }
229
230 pub fn chain(&self) -> Chain {
231 self.chain
232 }
233
234 pub fn tip_height(&self) -> Result<u64, Error> {
235 Ok(self.tip()?.0)
236 }
237
238 pub fn block_hash(&self, height: u64) -> Result<bitcoin::BlockHash, Error> {
239 let value = self.call("getblockhash", &[json!(height)])?;
240 Ok(value.as_str().context("missing block hash")?.parse()?)
241 }
242
243 pub fn block(&self, height: u64) -> Result<bitcoin::Block, Error> {
244 Ok(self.block_at(height)?.0)
245 }
246
247 pub fn block_at(&self, height: u64) -> Result<(bitcoin::Block, bitcoin::BlockHash), Error> {
248 let hash = self.block_hash(height)?;
249 let args = [json!(hash), json!(0)];
250 let (value, encoding) = match &self.backend {
251 Backend::Local(client) => (client.call("getblock", &args)?, BlockEncoding::Core),
252 Backend::Routed(router) => {
253 let answer = router.answer(self.chain, "getblock", &args)?;
254 tracing::debug!(provider = %answer.label, height, "block served by provider");
255 (answer.value, answer.encoding)
256 }
257 };
258 let raw = hex::decode(value.as_str().context("missing block bytes")?)?;
259 let decoded = match encoding {
260 BlockEncoding::Core => chain_decode::block(&raw, self.chain, hash),
261 BlockEncoding::Esplora => chain_decode::esplora_block(&raw, self.chain, hash),
262 };
263 let block = decoded.map_err(|cause| match cause {
264 DecodeError::Integrity(BlockValidationError::HashMismatch) => {
265 Error::Invalid("RPC block hash mismatch".into())
266 }
267 DecodeError::Integrity(BlockValidationError::MerkleRootMismatch) => {
268 Error::Invalid("RPC block merkle root mismatch".into())
269 }
270 cause => Error::from(cause),
271 })?;
272 Ok((block, hash))
273 }
274
275 pub fn confirmations(&self, txid: Txid) -> Result<u32, Error> {
276 match self.presence(txid)? {
277 Presence::Confirmed { height, .. } => Ok(u32::try_from(
278 self.tip_height()?
279 .checked_sub(height)
280 .context("chain changed during confirmation lookup")?
281 .checked_add(1)
282 .context("confirmation count overflow")?,
283 )?),
284 Presence::Missing | Presence::Mempool => Ok(0),
285 }
286 }
287
288 pub fn call(&self, method: &str, args: &[Value]) -> Result<Value, Error> {
289 match &self.backend {
290 Backend::Local(client) => Ok(client.call(method, args)?),
291 Backend::Routed(router) => {
292 if method != "sendrawtransaction" {
293 return router.call(self.chain, method, args);
294 }
295 let preference = self.parent_provider(args)?;
296 let answer = router.answer_preferring(&preference, self.chain, method, args)?;
297 self.remember_acknowledged(&answer.value, answer.label)?;
298 Ok(answer.value)
299 }
300 }
301 }
302
303 fn acknowledged(&self) -> MutexGuard<'_, HashMap<Txid, Acknowledged>> {
304 match self.acknowledged.lock() {
305 Ok(acknowledged) => acknowledged,
306 Err(poisoned) => {
307 tracing::error!("broadcast acknowledgment table lock poisoned");
308 poisoned.into_inner()
309 }
310 }
311 }
312
313 fn parent_provider(&self, args: &[Value]) -> Result<Preference, Error> {
314 let raw = args
315 .first()
316 .context("broadcast without a transaction")?
317 .as_str()
318 .context("broadcast transaction must be hex")?;
319 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
320 let acknowledged = self.acknowledged();
321 for input in &transaction.input {
322 let Some(parent) = acknowledged.get(&input.previous_output.txid) else {
323 continue;
324 };
325 if parent.when.elapsed() < config::SUBMISSION_ACK_WINDOW {
326 return Ok(Preference::Provider(parent.provider.clone()));
327 }
328 }
329 Ok(Preference::Any)
330 }
331
332 fn remember_acknowledged(&self, value: &Value, provider: String) -> Result<(), Error> {
333 let txid: Txid = value
334 .as_str()
335 .context("broadcast acknowledgment without a txid")?
336 .parse()?;
337 let now = Instant::now();
338 let mut acknowledged = self.acknowledged();
339 acknowledged.retain(|_txid, parent| {
340 now.duration_since(parent.when) < config::SUBMISSION_ACK_WINDOW
341 });
342 tracing::debug!(%txid, %provider, "broadcast acknowledged by a public provider");
343 acknowledged.insert(
344 txid,
345 Acknowledged {
346 when: now,
347 provider,
348 },
349 );
350 Ok(())
351 }
352
353 fn recently_acknowledged(&self, txid: Txid) -> bool {
354 let acknowledged = self.acknowledged();
355 let Some(parent) = acknowledged.get(&txid) else {
356 return false;
357 };
358 parent.when.elapsed() < config::SUBMISSION_ACK_WINDOW
359 }
360
361 pub fn presence_after_submission(&self, txid: Txid) -> Result<Presence, Error> {
362 let mut checks = 0;
363 loop {
364 checks += 1;
365 let presence = self.presence(txid)?;
366 if !matches!(presence, Presence::Missing)
367 || checks >= config::SUBMISSION_PRESENCE_CHECKS
368 {
369 return Ok(presence);
370 }
371 tracing::warn!(%txid, checks, "submitted transaction not yet visible; checking again");
372 std::thread::sleep(config::SUBMISSION_PRESENCE_DELAY);
373 }
374 }
375
376 pub fn require_txindex(&self) -> Result<(), Error> {
377 if matches!(self.backend, Backend::Routed(_)) {
378 return self.verify_network();
379 }
380 let indexes = self.call("getindexinfo", &[json!("txindex")])?;
381 ensure!(
382 indexes["txindex"]["synced"] == true,
383 "publication and recovery require synchronized txindex=1 on the local node"
384 );
385 Ok(())
386 }
387
388 pub fn verify_network(&self) -> Result<(), Error> {
389 let started = Instant::now();
390 let verified = (|| {
391 let genesis = self.call("getblockhash", &[json!(0)])?;
392 ensure!(
393 genesis.as_str() == Some(self.chain().genesis()?.0.to_string().as_str()),
394 "RPC genesis does not match selected chain"
395 );
396 Ok(())
397 })();
398 tracing::debug!(
399 target: "urma_startup",
400 chain = self.chain.label(),
401 stage = "network_verification",
402 elapsed_ms = started.elapsed().as_secs_f64() * 1000.0,
403 routed = self.is_public(),
404 result = ?verified,
405 "light client stage completed"
406 );
407 verified
408 }
409
410 pub fn tip(&self) -> Result<(u64, String), Error> {
411 let info = self.call("getblockchaininfo", &[])?;
412 Ok((
413 info["blocks"].as_u64().context("missing height")?,
414 info["bestblockhash"]
415 .as_str()
416 .context("missing tip hash")?
417 .to_owned(),
418 ))
419 }
420
421 pub fn transaction(&self, txid: Txid) -> Result<Transaction, Error> {
422 let raw = self.call("getrawtransaction", &[json!(txid), json!(false)])?;
423 self.decode_transaction(txid, &raw)
424 }
425
426 pub fn transaction_avoiding(
427 &self,
428 txid: Txid,
429 avoid: &[String],
430 ) -> Result<(Transaction, String), Error> {
431 let args = [json!(txid), json!(false)];
432 let (raw, provider) = match &self.backend {
433 Backend::Local(client) => {
434 if avoid
435 .iter()
436 .any(|label| label == config::LOCAL_PROVIDER_LABEL)
437 {
438 return Err(Error::Missing(format!(
439 "{txid}: the local node served a rejected candidate"
440 )));
441 }
442 (
443 client.call("getrawtransaction", &args)?,
444 config::LOCAL_PROVIDER_LABEL.to_owned(),
445 )
446 }
447 Backend::Routed(router) => {
448 let answer =
449 router.answer_avoiding(avoid, self.chain, "getrawtransaction", &args)?;
450 (answer.value, answer.label)
451 }
452 };
453 Ok((self.decode_transaction(txid, &raw)?, provider))
454 }
455
456 fn decode_transaction(&self, txid: Txid, raw: &Value) -> Result<Transaction, Error> {
457 let raw = raw.as_str().context("missing raw transaction")?;
458 let transaction = chain_decode::transaction(&hex::decode(raw)?, self.chain)?;
459 ensure!(
460 transaction.compute_txid() == txid,
461 "RPC transaction hash mismatch"
462 );
463 Ok(transaction)
464 }
465
466 pub fn presence(&self, txid: Txid) -> Result<Presence, Error> {
467 self.presence_cached(txid, &mut std::collections::HashMap::new())
468 }
469
470 pub(crate) fn presence_cached(
471 &self,
472 txid: Txid,
473 blocks: &mut std::collections::HashMap<String, Presence>,
474 ) -> Result<Presence, Error> {
475 let result = self.call("getrawtransaction", &[json!(txid), json!(true)]);
476 let value = match result {
477 Ok(value) => value,
478 Err(error @ Error::Rpc(_)) if error.is_missing_transaction() => {
479 tracing::warn!(%error, "transaction absent from node");
480 return Ok(Presence::Missing);
481 }
482 Err(Error::Missing(message)) => {
483 if self.recently_acknowledged(txid) {
484 tracing::warn!(%message, %txid, "providers lag behind an acknowledged broadcast; treating as mempool");
485 return Ok(Presence::Mempool);
486 }
487 tracing::warn!(%message, "transaction absent from public sources");
488 return Ok(Presence::Missing);
489 }
490 Err(error) => return Err(error),
491 };
492 if config::confirmations(&value)? <= 0 {
493 return Ok(Presence::Mempool);
494 }
495 let hash = value["blockhash"]
496 .as_str()
497 .context("missing inclusion block")?;
498 let slot = match blocks.entry(hash.to_owned()) {
499 Entry::Occupied(cached) => return Ok(cached.get().clone()),
500 Entry::Vacant(slot) => slot,
501 };
502 let header = self.call("getblockheader", &[json!(hash)])?;
503 let height = header["height"]
504 .as_u64()
505 .context("missing inclusion height")?;
506 let active = self.call("getblockhash", &[json!(height)])?;
507 if active != json!(hash) {
508 return Ok(Presence::Missing);
509 }
510 let presence = Presence::Confirmed {
511 height,
512 block_hash: hash.to_owned(),
513 };
514 slot.insert(presence.clone());
515 Ok(presence)
516 }
517}