1use crate::config;
2use crate::error::{Error, ensure};
3use crate::remote::validate_result;
4use serde_json::Value;
5use std::{
6 num::Saturating,
7 sync::{Mutex, mpsc},
8 time::{Duration, Instant},
9};
10use urma_chain::observation::Chain;
11
12#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
13pub enum Evidence {
14 LocalValidatingNode,
15 LightClientInclusion,
16 PublicProviderObservation,
17}
18
19impl Evidence {
20 pub fn label(self) -> &'static str {
21 match self {
22 Self::LocalValidatingNode => "local_validating_node",
23 Self::LightClientInclusion => "light_client_inclusion",
24 Self::PublicProviderObservation => "public_provider_observation",
25 }
26 }
27}
28
29#[derive(Clone, Copy, Debug, PartialEq, Eq)]
30pub enum BlockEncoding {
31 Core,
32 Esplora,
33}
34
35pub trait Provider: Send + Sync {
36 fn label(&self) -> String;
37 fn evidence(&self) -> Evidence;
38 fn block_encoding(&self) -> BlockEncoding;
39 fn supports(&self, method: &str) -> bool;
40 fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error>;
41}
42
43pub struct Answer {
44 pub value: Value,
45 pub label: String,
46 pub encoding: BlockEncoding,
47 pub evidence: Evidence,
48}
49
50#[derive(Clone, Debug, PartialEq, Eq)]
51pub enum Preference {
52 Any,
53 Provider(String),
54}
55
56#[derive(Clone, Debug, PartialEq, Eq)]
57pub struct Standing {
58 pub label: String,
59 pub evidence: Evidence,
60 pub failures: u32,
61 pub blocked_for: Duration,
62}
63
64enum Outcome {
65 Answered(Answer),
66 Absent(String),
67 Failed(String),
68}
69
70pub struct Router {
71 slots: Vec<Slot>,
72 race_tip: bool,
73}
74
75struct Slot {
76 provider: Box<dyn Provider>,
77 health: Mutex<Health>,
78}
79
80#[derive(Clone, Copy)]
81struct Health {
82 failures: Saturating<u32>,
83 blocked_until: Option<Instant>,
84}
85
86impl Slot {
87 fn health(&self) -> Health {
88 match self.health.lock() {
89 Ok(health) => *health,
90 Err(poisoned) => {
91 tracing::error!(provider = %self.provider.label(), "provider health lock poisoned");
92 *poisoned.into_inner()
93 }
94 }
95 }
96
97 fn update(&self, change: impl FnOnce(&mut Health)) {
98 match self.health.lock() {
99 Ok(mut health) => change(&mut health),
100 Err(poisoned) => {
101 tracing::error!(provider = %self.provider.label(), "provider health lock poisoned");
102 change(&mut poisoned.into_inner())
103 }
104 }
105 }
106
107 fn succeeded(&self) {
108 self.update(|health| {
109 health.failures = Saturating(0);
110 health.blocked_until = None;
111 });
112 }
113
114 fn failed(&self, now: Instant) {
115 self.update(|health| {
116 health.failures += Saturating(1);
117 if health.failures.0 >= config::PROVIDER_BREAKER_FAILURES {
118 health.blocked_until = Some(now + config::PROVIDER_BREAKER_WINDOW);
119 tracing::warn!(provider = %self.provider.label(), failures = health.failures.0, "provider circuit opened");
120 }
121 });
122 }
123
124 fn cooled(&self, now: Instant, wait: Duration) {
125 self.update(|health| {
126 health.blocked_until = Some(now + wait.min(config::PROVIDER_RETRY_MAX));
127 });
128 }
129}
130
131impl Router {
132 pub fn new(providers: Vec<Box<dyn Provider>>) -> Result<Self, Error> {
133 ensure!(
134 !providers.is_empty() && providers.len() <= config::MAX_PROVIDERS,
135 "public transport requires between one and {} providers",
136 config::MAX_PROVIDERS
137 );
138 let mut slots = Vec::new();
139 for provider in providers {
140 slots.push(Slot {
141 provider,
142 health: Mutex::new(Health {
143 failures: Saturating(0),
144 blocked_until: None,
145 }),
146 });
147 }
148 Ok(Self {
149 slots,
150 race_tip: false,
151 })
152 }
153
154 pub fn race_tip(mut self, enabled: bool) -> Self {
155 self.race_tip = enabled;
156 self
157 }
158
159 pub fn evidence(&self) -> Evidence {
160 let mut weakest = Evidence::LightClientInclusion;
161 for slot in &self.slots {
162 weakest = weakest.max(slot.provider.evidence());
163 }
164 weakest
165 }
166
167 pub fn labels(&self) -> Vec<String> {
168 self.slots
169 .iter()
170 .map(|slot| slot.provider.label())
171 .collect()
172 }
173
174 pub fn standings(&self) -> Vec<Standing> {
175 let now = Instant::now();
176 self.slots
177 .iter()
178 .map(|slot| {
179 let health = slot.health();
180 Standing {
181 label: slot.provider.label(),
182 evidence: slot.provider.evidence(),
183 failures: health.failures.0,
184 blocked_for: health
185 .blocked_until
186 .and_then(|deadline| deadline.checked_duration_since(now))
187 .unwrap_or(Duration::ZERO),
188 }
189 })
190 .collect()
191 }
192
193 pub fn call(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Value, Error> {
194 Ok(self.answer(chain, method, args)?.value)
195 }
196
197 pub fn answer(&self, chain: Chain, method: &str, args: &[Value]) -> Result<Answer, Error> {
198 self.answer_preferring(&Preference::Any, chain, method, args)
199 }
200
201 pub fn answer_preferring(
202 &self,
203 preference: &Preference,
204 chain: Chain,
205 method: &str,
206 args: &[Value],
207 ) -> Result<Answer, Error> {
208 let mut candidates = self.candidates(method, Instant::now())?;
209 match preference {
210 Preference::Provider(label) => {
211 let found = candidates
212 .iter()
213 .position(|index| self.slots[*index].provider.label() == *label);
214 for position in found.iter() {
215 let index = candidates.remove(*position);
216 candidates.insert(0, index);
217 tracing::debug!(provider = %label, method, "preferring the provider that acknowledged the parent");
218 }
219 }
220 Preference::Any => (),
221 }
222 self.answer_from(&candidates, chain, method, args)
223 }
224
225 pub fn answer_avoiding(
226 &self,
227 avoid: &[String],
228 chain: Chain,
229 method: &str,
230 args: &[Value],
231 ) -> Result<Answer, Error> {
232 let mut candidates = self.candidates(method, Instant::now())?;
233 candidates.retain(|index| !avoid.contains(&self.slots[*index].provider.label()));
234 if candidates.is_empty()
235 && self.slots.iter().all(|slot| {
236 !slot.provider.supports(method) || avoid.contains(&slot.provider.label())
237 })
238 {
239 return Err(Error::Missing(format!(
240 "{method} on {chain:?}: every provider that serves it was rejected"
241 )));
242 }
243 self.answer_from(&candidates, chain, method, args)
244 }
245
246 fn answer_from(
247 &self,
248 candidates: &[usize],
249 chain: Chain,
250 method: &str,
251 args: &[Value],
252 ) -> Result<Answer, Error> {
253 ensure!(
254 !candidates.is_empty(),
255 "{method} on {chain:?}: no public provider is available for this method right now; retry shortly or configure a local node"
256 );
257 if self.race_tip && method == "getblockchaininfo" && candidates.len() > 1 {
258 let width = candidates.len().min(config::TIP_RACE_WIDTH);
259 return self.race(chain, &candidates[..width], method, args);
260 }
261 let mut failures = Vec::new();
262 let mut absent = 0;
263 for index in candidates {
264 match self.attempt(&self.slots[*index], chain, method, args) {
265 Outcome::Answered(answer) => return Ok(answer),
266 Outcome::Absent(message) => {
267 absent += 1;
268 failures.push(message);
269 }
270 Outcome::Failed(message) => failures.push(message),
271 }
272 }
273 let message = format!(
274 "{method} on {chain:?}: {}. Retry or configure a local node; check --testnet for test data.",
275 failures.join("; ")
276 );
277 if absent > 0 {
278 if absent < candidates.len() {
279 tracing::warn!(
280 method,
281 absent,
282 failed = candidates.len() - absent,
283 %message,
284 "record absent from every provider that answered; other providers failed"
285 );
286 }
287 return Err(Error::Missing(message));
288 }
289 Err(Error::Unsupported(message))
290 }
291
292 fn candidates(&self, method: &str, now: Instant) -> Result<Vec<usize>, Error> {
293 ensure!(
294 config::PUBLIC_METHODS.contains(&method),
295 "public transport supports chain reads and signed transaction submission only"
296 );
297 let mut ranked = Vec::new();
298 for (index, slot) in self.slots.iter().enumerate() {
299 if !slot.provider.supports(method) {
300 continue;
301 }
302 let health = slot.health();
303 if health.blocked_until.is_some_and(|deadline| deadline > now) {
304 continue;
305 }
306 ranked.push((slot.provider.evidence(), health.failures.0, index));
307 }
308 ranked.sort();
309 let mut order = Vec::new();
310 for (evidence, failures, index) in ranked {
311 tracing::debug!(?evidence, failures, index, method, "provider candidate");
312 order.push(index);
313 }
314 Ok(order)
315 }
316
317 fn attempt(&self, slot: &Slot, chain: Chain, method: &str, args: &[Value]) -> Outcome {
318 let label = slot.provider.label();
319 let encoding = slot.provider.block_encoding();
320 let evidence = slot.provider.evidence();
321 let result = slot
322 .provider
323 .call(chain, method, args)
324 .and_then(|value| validate_result(encoding, chain, method, args, value));
325 match result {
326 Ok(value) => {
327 slot.succeeded();
328 Outcome::Answered(Answer {
329 value,
330 label,
331 encoding,
332 evidence,
333 })
334 }
335 Err(Error::Missing(message)) if absent_record(&message) => {
336 slot.succeeded();
337 tracing::warn!(provider = %label, %message, "record absent from one provider");
338 Outcome::Absent(message)
339 }
340 Err(Error::RateLimited(wait)) => {
341 slot.cooled(Instant::now(), wait);
342 tracing::warn!(provider = %label, seconds = wait.as_secs(), "provider rate limited");
343 Outcome::Failed(Error::RateLimited(wait).to_string())
344 }
345 Err(error) => {
346 slot.failed(Instant::now());
347 tracing::warn!(provider = %label, %error, "public provider unavailable");
348 Outcome::Failed(error.to_string())
349 }
350 }
351 }
352
353 fn race(
354 &self,
355 chain: Chain,
356 indices: &[usize],
357 method: &str,
358 args: &[Value],
359 ) -> Result<Answer, Error> {
360 let (sender, receiver) = mpsc::sync_channel(indices.len());
361 std::thread::scope(|scope| {
362 for index in indices {
363 let sender = sender.clone();
364 let slot = &self.slots[*index];
365 scope.spawn(move || match sender.send(self.attempt(slot, chain, method, args)) {
366 Ok(()) => (),
367 Err(error) => {
368 tracing::warn!(%error, "tip race answer completed after another provider won")
369 }
370 });
371 }
372 drop(sender);
373 let mut failures = Vec::new();
374 for outcome in receiver {
375 match outcome {
376 Outcome::Answered(answer) => return Ok(answer),
377 Outcome::Absent(message) | Outcome::Failed(message) => failures.push(message),
378 }
379 }
380 Err(Error::Unsupported(format!(
381 "{method} on {chain:?}: {}",
382 failures.join("; ")
383 )))
384 })
385 }
386}
387
388fn absent_record(message: &str) -> bool {
389 message == config::ABSENT_TRANSACTION || message == config::ABSENT_RECORD
390}
391
392#[cfg(test)]
393mod health_tests {
394 use super::*;
395 use serde_json::json;
396
397 struct Ready;
398 impl Provider for Ready {
399 fn label(&self) -> String {
400 "ready fixture".into()
401 }
402 fn evidence(&self) -> Evidence {
403 Evidence::PublicProviderObservation
404 }
405 fn block_encoding(&self) -> BlockEncoding {
406 BlockEncoding::Core
407 }
408 fn supports(&self, method: &str) -> bool {
409 method == "getblockhash"
410 }
411 fn call(&self, _: Chain, _: &str, _: &[Value]) -> Result<Value, Error> {
412 Ok(json!("00".repeat(32)))
413 }
414 }
415
416 #[test]
417 fn success_after_candidate_clock_sample_does_not_cool_healthy_provider() {
418 let router = Router::new(vec![Box::new(Ready)]).unwrap();
419 let query_started = Instant::now();
422 router
423 .call(Chain::BitcoinRegtest, "getblockhash", &[json!(1)])
424 .unwrap();
425 assert_eq!(
426 router.candidates("getblockhash", query_started).unwrap(),
427 vec![0]
428 );
429 }
430
431 #[test]
432 fn real_cooldown_still_blocks_until_deadline_and_success_clears_it() {
433 let router = Router::new(vec![Box::new(Ready)]).unwrap();
434 let now = Instant::now();
435 router.slots[0].cooled(now, Duration::from_secs(5));
436 assert!(router.candidates("getblockhash", now).unwrap().is_empty());
437 assert_eq!(
438 router
439 .candidates("getblockhash", now + Duration::from_secs(5))
440 .unwrap(),
441 vec![0]
442 );
443 router.slots[0].succeeded();
444 assert_eq!(
445 router.candidates("getblockhash", Instant::now()).unwrap(),
446 vec![0]
447 );
448 }
449}