1#![forbid(unsafe_code)]
10
11use std::future::Future;
12use std::pin::Pin;
13use std::sync::Arc;
14use std::time::{SystemTime, UNIX_EPOCH};
15
16use chio_core::web3::anchors::{
17 OracleConversionEvidence, CHIO_LINK_ORACLE_AUTHORITY, CHIO_ORACLE_CONVERSION_EVIDENCE_SCHEMA,
18};
19use tokio::sync::RwLock;
20
21pub mod cache;
22#[cfg_attr(feature = "web3", path = "chainlink.rs")]
23#[cfg_attr(not(feature = "web3"), path = "chainlink_disabled.rs")]
24pub mod chainlink;
25pub mod circuit_breaker;
26pub mod config;
27pub mod control;
28pub mod convert;
29pub mod monitor;
30pub mod pyth;
31mod reports;
32#[cfg_attr(feature = "web3", path = "sequencer.rs")]
33#[cfg_attr(not(feature = "web3"), path = "sequencer_disabled.rs")]
34pub mod sequencer;
35
36#[cfg(test)]
37pub(crate) mod test_support {
38 pub(crate) use chio_test_support::ctx::{TestUnwrap, TestUnwrapErr};
39}
40
41use cache::PriceCache;
42#[cfg(feature = "web3")]
43use chainlink::ChainlinkFeedReader;
44use circuit_breaker::ensure_within_threshold;
45use config::{
46 pair_key, OperatorConfig, OracleBackendKind, PairConfig, PairRuntimeOverride, PriceOracleConfig,
47};
48use monitor::{
49 AlertSeverity, ChainHealthReport, ChainHealthStatus, OracleAlert, OracleRuntimeReport,
50 PairHealthReport,
51};
52use pyth::PythHermesClient;
53use reports::{
54 alert_for_chain, alert_for_pair, classify_error_status, classify_success_status,
55 confidence_bps, pair_success_note,
56};
57use sequencer::{read_sequencer_status, SequencerAvailability};
58
59pub type OracleFuture<'a> =
60 Pin<Box<dyn Future<Output = Result<ExchangeRate, PriceOracleError>> + Send + 'a>>;
61
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct ExchangeRate {
64 pub base: String,
65 pub quote: String,
66 pub rate_numerator: u128,
67 pub rate_denominator: u128,
68 pub updated_at: u64,
69 pub fetched_at: u64,
70 pub source: String,
71 pub feed_reference: String,
72 pub max_age_seconds: u64,
73 pub conversion_margin_bps: u32,
74 pub confidence_numerator: Option<u128>,
75 pub confidence_denominator: Option<u128>,
76}
77
78impl ExchangeRate {
79 #[must_use]
80 pub fn pair(&self) -> String {
81 pair_key(&self.base, &self.quote)
82 }
83
84 pub fn ensure_matches_pair(&self, pair: &PairConfig) -> Result<(), PriceOracleError> {
85 if self.base != pair.base || self.quote != pair.quote {
86 return Err(PriceOracleError::InvalidFeed(format!(
87 "oracle backend returned {}/{} for {}",
88 self.base,
89 self.quote,
90 pair.pair()
91 )));
92 }
93 Ok(())
94 }
95
96 #[must_use]
97 pub fn age_seconds(&self, now: u64) -> u64 {
98 now.saturating_sub(self.updated_at)
99 }
100
101 #[must_use]
102 pub fn cache_age_seconds(&self, now: u64) -> u64 {
103 now.saturating_sub(self.fetched_at)
104 }
105
106 pub fn ensure_fresh(&self, now: u64) -> Result<(), PriceOracleError> {
107 if self.rate_denominator == 0 {
108 return Err(PriceOracleError::InvalidFeed(format!(
109 "{} returned a zero rate denominator",
110 self.pair()
111 )));
112 }
113 if self.updated_at > now {
114 return Err(PriceOracleError::InvalidFeed(format!(
115 "{} returned a future updated_at timestamp",
116 self.pair()
117 )));
118 }
119 let age_seconds = self.age_seconds(now);
120 if age_seconds > self.max_age_seconds {
121 return Err(PriceOracleError::Stale {
122 pair: self.pair(),
123 age_seconds,
124 max_age_seconds: self.max_age_seconds,
125 });
126 }
127 Ok(())
128 }
129
130 #[must_use]
131 pub fn backend_label(&self) -> String {
132 self.source
133 .split(':')
134 .next()
135 .unwrap_or(self.source.as_str())
136 .to_string()
137 }
138
139 #[must_use]
140 pub fn with_degraded_mode(
141 mut self,
142 max_age_seconds: u64,
143 extra_margin_bps: u32,
144 source_suffix: &str,
145 ) -> Self {
146 self.max_age_seconds = max_age_seconds;
147 self.conversion_margin_bps = self.conversion_margin_bps.saturating_add(extra_margin_bps);
148 self.source = format!("{}:{source_suffix}", self.source);
149 self
150 }
151
152 pub fn to_conversion_evidence(
153 &self,
154 original_cost_units: u64,
155 original_currency: impl Into<String>,
156 grant_currency: impl Into<String>,
157 converted_cost_units: u64,
158 now: u64,
159 ) -> Result<OracleConversionEvidence, PriceOracleError> {
160 self.ensure_fresh(now)?;
161 Ok(OracleConversionEvidence {
162 schema: CHIO_ORACLE_CONVERSION_EVIDENCE_SCHEMA.to_string(),
163 base: self.base.clone(),
164 quote: self.quote.clone(),
165 authority: CHIO_LINK_ORACLE_AUTHORITY.to_string(),
166 rate_numerator: u64::try_from(self.rate_numerator).map_err(|_| {
167 PriceOracleError::ArithmeticOverflow(format!(
168 "{} rate_numerator does not fit the receipt contract",
169 self.pair()
170 ))
171 })?,
172 rate_denominator: u64::try_from(self.rate_denominator).map_err(|_| {
173 PriceOracleError::ArithmeticOverflow(format!(
174 "{} rate_denominator does not fit the receipt contract",
175 self.pair()
176 ))
177 })?,
178 source: self.source.clone(),
179 feed_address: self.feed_reference.clone(),
180 updated_at: self.updated_at,
181 max_age_seconds: self.max_age_seconds,
182 cache_age_seconds: self.cache_age_seconds(now),
183 converted_cost_units,
184 original_cost_units,
185 original_currency: original_currency.into(),
186 grant_currency: grant_currency.into(),
187 oracle_public_key: None,
188 signature: None,
189 })
190 }
191}
192
193#[derive(Debug, thiserror::Error, Clone)]
194pub enum PriceOracleError {
195 #[error("no feed configured for {base}/{quote}")]
196 NoPairAvailable { base: String, quote: String },
197 #[error("invalid price oracle configuration: {0}")]
198 InvalidConfiguration(String),
199 #[error("unsupported oracle backend: {0}")]
200 UnsupportedBackend(String),
201 #[error("price stale: age {age_seconds}s exceeds max {max_age_seconds}s for {pair}")]
202 Stale {
203 pair: String,
204 age_seconds: u64,
205 max_age_seconds: u64,
206 },
207 #[error(
208 "oracle sources diverge by {divergence_pct:.2}% for {pair} (threshold: {threshold_pct:.2}%)"
209 )]
210 CircuitBreakerTripped {
211 pair: String,
212 divergence_pct: f64,
213 threshold_pct: f64,
214 },
215 #[error("operator pause is active{pair_suffix}: {reason}")]
216 OperatorPaused { pair_suffix: String, reason: String },
217 #[error("trusted chain {chain_id} is disabled for {pair}")]
218 ChainDisabled { pair: String, chain_id: u64 },
219 #[error("L2 sequencer is down for chain {chain_id} while resolving {pair} ({feed_address})")]
220 SequencerDown {
221 pair: String,
222 chain_id: u64,
223 feed_address: String,
224 },
225 #[error(
226 "L2 sequencer recovery grace remains active for chain {chain_id} while resolving {pair} ({feed_address}); {remaining_seconds}s remaining"
227 )]
228 SequencerRecovering {
229 pair: String,
230 chain_id: u64,
231 feed_address: String,
232 remaining_seconds: u64,
233 },
234 #[error("oracle backend unavailable: {0}")]
235 Unavailable(String),
236 #[error("invalid feed response: {0}")]
237 InvalidFeed(String),
238 #[error("arithmetic overflow: {0}")]
239 ArithmeticOverflow(String),
240}
241
242impl PriceOracleError {
243 pub(crate) fn invalid_configuration(message: impl Into<String>) -> Self {
244 Self::InvalidConfiguration(message.into())
245 }
246}
247
248pub trait PriceOracle: Send + Sync {
249 fn get_rate<'a>(&'a self, base: &'a str, quote: &'a str) -> OracleFuture<'a>;
250
251 fn supported_pairs(&self) -> Vec<String>;
252}
253
254pub trait OracleBackend: Send + Sync {
255 fn kind(&self) -> OracleBackendKind;
256
257 fn read_rate<'a>(&'a self, pair: &'a PairConfig, now: u64) -> OracleFuture<'a>;
258}
259
260pub struct ChioLinkOracle {
261 config: PriceOracleConfig,
262 primary: Arc<dyn OracleBackend>,
263 fallback: Option<Arc<dyn OracleBackend>>,
264 cache: RwLock<PriceCache>,
265 operator: RwLock<OperatorConfig>,
266}
267
268impl ChioLinkOracle {
269 pub fn new(config: PriceOracleConfig) -> Result<Self, PriceOracleError> {
270 config.validate()?;
271 let primary = build_backend(config.primary, &config)?;
272 let fallback = match config.fallback {
273 Some(kind) => Some(build_backend(kind, &config)?),
274 None => None,
275 };
276 Self::new_with_backends(config, primary, fallback)
277 }
278
279 pub fn new_with_backends(
280 config: PriceOracleConfig,
281 primary: Arc<dyn OracleBackend>,
282 fallback: Option<Arc<dyn OracleBackend>>,
283 ) -> Result<Self, PriceOracleError> {
284 config.validate()?;
285 if primary.kind() != config.primary {
286 return Err(PriceOracleError::InvalidConfiguration(format!(
287 "configured primary backend {:?} does not match implementation {:?}",
288 config.primary,
289 primary.kind()
290 )));
291 }
292 if let (Some(expected), Some(backend)) = (config.fallback, fallback.as_ref()) {
293 if backend.kind() != expected {
294 return Err(PriceOracleError::InvalidConfiguration(format!(
295 "configured fallback backend {:?} does not match implementation {:?}",
296 expected,
297 backend.kind()
298 )));
299 }
300 }
301 Ok(Self {
302 operator: RwLock::new(config.operator.clone()),
303 config,
304 primary,
305 fallback,
306 cache: RwLock::new(PriceCache::default()),
307 })
308 }
309
310 #[must_use]
311 pub fn config(&self) -> &PriceOracleConfig {
312 &self.config
313 }
314
315 pub async fn operator_config(&self) -> OperatorConfig {
316 self.operator.read().await.clone()
317 }
318
319 pub async fn set_global_pause(
320 &self,
321 paused: bool,
322 reason: Option<String>,
323 ) -> Result<(), PriceOracleError> {
324 let mut operator = self.operator.write().await;
325 operator.global_pause = paused;
326 operator.pause_reason = reason;
327 self.config.validate()?;
328 Ok(())
329 }
330
331 pub async fn set_chain_enabled(
332 &self,
333 chain_id: u64,
334 enabled: bool,
335 ) -> Result<(), PriceOracleError> {
336 let mut operator = self.operator.write().await;
337 let chain = operator.chain(chain_id).cloned().ok_or_else(|| {
338 PriceOracleError::InvalidConfiguration(format!(
339 "cannot toggle unknown operator chain_id {}",
340 chain_id
341 ))
342 })?;
343 let index = operator
344 .chains
345 .iter()
346 .position(|current| current.chain_id == chain.chain_id)
347 .ok_or_else(|| {
348 PriceOracleError::InvalidConfiguration(format!(
349 "operator chain_id {} vanished during update",
350 chain_id
351 ))
352 })?;
353 operator.chains[index].enabled = enabled;
354 Ok(())
355 }
356
357 pub async fn set_pair_override(
358 &self,
359 pair_override: PairRuntimeOverride,
360 ) -> Result<(), PriceOracleError> {
361 self.config
362 .pair(&pair_override.base, &pair_override.quote)
363 .ok_or_else(|| {
364 PriceOracleError::InvalidConfiguration(format!(
365 "cannot override unsupported pair {}",
366 pair_override.pair()
367 ))
368 })?;
369 let mut operator = self.operator.write().await;
370 let pair_key = pair_override.pair();
371 if let Some(index) = operator
372 .pair_overrides
373 .iter()
374 .position(|current| current.pair() == pair_key)
375 {
376 operator.pair_overrides[index] = pair_override;
377 } else {
378 operator.pair_overrides.push(pair_override);
379 }
380 Ok(())
381 }
382
383 pub async fn cached_rate(
384 &self,
385 base: &str,
386 quote: &str,
387 ) -> Result<Option<ExchangeRate>, PriceOracleError> {
388 let pair =
389 self.pair_config(base, quote)?
390 .ok_or_else(|| PriceOracleError::NoPairAvailable {
391 base: base.to_ascii_uppercase(),
392 quote: quote.to_ascii_uppercase(),
393 })?;
394 let now = now_unix()?;
395 let operator = self.operator_config().await;
396 self.resolve_cached_rate(&pair, now, &operator).await
397 }
398
399 pub async fn refresh_pair(
400 &self,
401 base: &str,
402 quote: &str,
403 ) -> Result<ExchangeRate, PriceOracleError> {
404 let pair =
405 self.pair_config(base, quote)?
406 .ok_or_else(|| PriceOracleError::NoPairAvailable {
407 base: base.to_ascii_uppercase(),
408 quote: quote.to_ascii_uppercase(),
409 })?;
410 let now = now_unix()?;
411 let operator = self.operator_config().await;
412 self.refresh_pair_inner(&pair, now, &operator).await
413 }
414
415 pub async fn runtime_report(&self) -> Result<OracleRuntimeReport, PriceOracleError> {
416 let now = now_unix()?;
417 let operator = self.operator_config().await;
418 let mut report = OracleRuntimeReport::new(now);
419 report.global_pause = operator.global_pause;
420 report.pause_reason = operator.pause_reason.clone();
421
422 for chain in &operator.chains {
423 let chain_report = self.chain_health_report(chain, now).await;
424 if operator.monitoring.alert_on_sequencer {
425 if let Some(alert) = alert_for_chain(&chain_report, now) {
426 report.alerts.push(alert);
427 }
428 }
429 report.chains.push(chain_report);
430 }
431
432 for pair in &self.config.pairs {
433 let pair_report = self.pair_health_report(pair, &operator, now).await;
434 if let Some(alert) = alert_for_pair(&pair_report, &operator.monitoring, now) {
435 report.alerts.push(alert);
436 }
437 report.pairs.push(pair_report);
438 }
439
440 if operator.global_pause && operator.monitoring.alert_on_pause {
441 report.alerts.push(OracleAlert {
442 code: "global_pause".to_string(),
443 severity: AlertSeverity::Critical,
444 message: operator
445 .pause_reason
446 .clone()
447 .unwrap_or_else(|| "chio-link global pause is active".to_string()),
448 pair: None,
449 chain_id: None,
450 observed_at: now,
451 });
452 }
453
454 Ok(report)
455 }
456
457 async fn chain_health_report(
458 &self,
459 chain: &config::ChainlinkNetworkConfig,
460 now: u64,
461 ) -> ChainHealthReport {
462 if !chain.enabled {
463 return ChainHealthReport {
464 chain_id: chain.chain_id,
465 label: chain.label.clone(),
466 caip2: chain.caip2.clone(),
467 enabled: false,
468 status: ChainHealthStatus::Disabled,
469 sequencer_uptime_feed: chain.sequencer_uptime_feed.clone(),
470 checked_at: Some(now),
471 status_started_at: None,
472 note: Some("operator disabled this chain".to_string()),
473 };
474 }
475
476 match read_sequencer_status(chain, now, &self.config.egress_contract).await {
477 Ok(Some(status)) => {
478 let (health, note) = match status.availability {
479 SequencerAvailability::Up => (ChainHealthStatus::Healthy, None),
480 SequencerAvailability::Down => (
481 ChainHealthStatus::Down,
482 Some("L2 sequencer is down; fail closed".to_string()),
483 ),
484 SequencerAvailability::Recovering => (
485 ChainHealthStatus::Recovering,
486 Some("L2 sequencer recently recovered; grace window active".to_string()),
487 ),
488 };
489 ChainHealthReport {
490 chain_id: chain.chain_id,
491 label: chain.label.clone(),
492 caip2: chain.caip2.clone(),
493 enabled: true,
494 status: health,
495 sequencer_uptime_feed: Some(status.feed_address),
496 checked_at: Some(status.checked_at),
497 status_started_at: Some(status.status_started_at),
498 note,
499 }
500 }
501 Ok(None) => ChainHealthReport {
502 chain_id: chain.chain_id,
503 label: chain.label.clone(),
504 caip2: chain.caip2.clone(),
505 enabled: true,
506 status: ChainHealthStatus::Unmonitored,
507 sequencer_uptime_feed: None,
508 checked_at: Some(now),
509 status_started_at: None,
510 note: Some("no sequencer uptime feed configured".to_string()),
511 },
512 Err(error) => ChainHealthReport {
513 chain_id: chain.chain_id,
514 label: chain.label.clone(),
515 caip2: chain.caip2.clone(),
516 enabled: true,
517 status: ChainHealthStatus::Unavailable,
518 sequencer_uptime_feed: chain.sequencer_uptime_feed.clone(),
519 checked_at: Some(now),
520 status_started_at: None,
521 note: Some(error.to_string()),
522 },
523 }
524 }
525
526 async fn pair_health_report(
527 &self,
528 pair: &PairConfig,
529 operator: &OperatorConfig,
530 now: u64,
531 ) -> PairHealthReport {
532 let pair_override = self.effective_pair_override(operator, pair);
533 match self.rate_for_pair(pair, now, operator).await {
534 Ok(rate) => {
535 let pair_status =
536 classify_success_status(&rate, &pair_override, self.config.primary);
537 PairHealthReport {
538 pair: pair.pair(),
539 chain_id: pair.chain_id,
540 status: pair_status,
541 active_backend: Some(rate.backend_label()),
542 active_source: Some(rate.source.clone()),
543 feed_reference: Some(rate.feed_reference.clone()),
544 updated_at: Some(rate.updated_at),
545 cache_age_seconds: Some(rate.cache_age_seconds(now)),
546 conversion_margin_bps: Some(rate.conversion_margin_bps),
547 confidence_bps: confidence_bps(&rate),
548 note: pair_success_note(&rate, &pair_override),
549 last_error: None,
550 }
551 }
552 Err(error) => PairHealthReport {
553 pair: pair.pair(),
554 chain_id: pair.chain_id,
555 status: classify_error_status(&error),
556 active_backend: None,
557 active_source: None,
558 feed_reference: None,
559 updated_at: None,
560 cache_age_seconds: None,
561 conversion_margin_bps: None,
562 confidence_bps: None,
563 note: None,
564 last_error: Some(error.to_string()),
565 },
566 }
567 }
568
569 async fn rate_for_pair(
570 &self,
571 pair: &PairConfig,
572 now: u64,
573 operator: &OperatorConfig,
574 ) -> Result<ExchangeRate, PriceOracleError> {
575 if let Some(rate) = self.resolve_cached_rate(pair, now, operator).await? {
576 return Ok(rate);
577 }
578 self.refresh_pair_inner(pair, now, operator).await
579 }
580
581 async fn resolve_cached_rate(
582 &self,
583 pair: &PairConfig,
584 now: u64,
585 operator: &OperatorConfig,
586 ) -> Result<Option<ExchangeRate>, PriceOracleError> {
587 self.enforce_operator_controls(pair, now, operator).await?;
588 let pair_override = self.effective_pair_override(operator, pair);
589 let mut cache = self.cache.write().await;
590 match cache.resolve(pair, now) {
591 Ok(rate) => Ok(rate),
592 Err(error @ PriceOracleError::Stale { .. }) => {
593 if let Some(stale_rate) = cache.latest(pair) {
594 if let Some(rate) =
595 degraded_rate_if_allowed(pair, &pair_override, stale_rate, now)
596 {
597 return Ok(Some(rate));
598 }
599 }
600 Err(error)
601 }
602 Err(error) => Err(error),
603 }
604 }
605
606 async fn refresh_pair_inner(
607 &self,
608 pair: &PairConfig,
609 now: u64,
610 operator: &OperatorConfig,
611 ) -> Result<ExchangeRate, PriceOracleError> {
612 self.enforce_operator_controls(pair, now, operator).await?;
613 let rate = self.fetch_authoritative_rate(pair, now, operator).await?;
614 let mut cache = self.cache.write().await;
615 cache.record(pair, rate, now)?;
616 cache.resolve(pair, now)?.ok_or_else(|| {
617 PriceOracleError::Unavailable(format!(
618 "cache entry missing after refresh for {}",
619 pair.pair()
620 ))
621 })
622 }
623
624 async fn fetch_authoritative_rate(
625 &self,
626 pair: &PairConfig,
627 now: u64,
628 operator: &OperatorConfig,
629 ) -> Result<ExchangeRate, PriceOracleError> {
630 let pair_override = self.effective_pair_override(operator, pair);
631
632 if let Some(kind) = pair_override.force_backend {
633 let backend = self.backend_for_kind(kind).ok_or_else(|| {
634 PriceOracleError::UnsupportedBackend(format!(
635 "forced backend {:?} is not available in chio-link",
636 kind
637 ))
638 })?;
639 if !pair_supports_backend(pair, kind) {
640 return Err(PriceOracleError::UnsupportedBackend(format!(
641 "{} does not support forced backend {:?}",
642 pair.pair(),
643 kind
644 )));
645 }
646 return read_validated_backend_rate(backend.as_ref(), pair, now).await;
647 }
648
649 match read_validated_backend_rate(self.primary.as_ref(), pair, now).await {
650 Ok(primary_rate) => {
651 if let Some(secondary) = self.secondary_backend_for_pair(pair, &pair_override) {
652 if let Ok(secondary_rate) =
653 read_validated_backend_rate(secondary.as_ref(), pair, now).await
654 {
655 ensure_within_threshold(
656 &primary_rate,
657 &secondary_rate,
658 pair_override
659 .divergence_threshold_bps
660 .unwrap_or(pair.policy.divergence_threshold_bps),
661 )?;
662 }
663 }
664 Ok(primary_rate)
665 }
666 Err(primary_error) => {
667 if let Some(secondary) = self.secondary_backend_for_pair(pair, &pair_override) {
668 read_validated_backend_rate(secondary.as_ref(), pair, now).await
669 } else {
670 Err(primary_error)
671 }
672 }
673 }
674 }
675
676 async fn enforce_operator_controls(
677 &self,
678 pair: &PairConfig,
679 now: u64,
680 operator: &OperatorConfig,
681 ) -> Result<(), PriceOracleError> {
682 if operator.global_pause {
683 return Err(PriceOracleError::OperatorPaused {
684 pair_suffix: format!(" for {}", pair.pair()),
685 reason: operator
686 .pause_reason
687 .clone()
688 .unwrap_or_else(|| "operator global pause is active".to_string()),
689 });
690 }
691
692 let pair_override = self.effective_pair_override(operator, pair);
693 if !pair_override.enabled {
694 return Err(PriceOracleError::OperatorPaused {
695 pair_suffix: format!(" for {}", pair.pair()),
696 reason: "pair-specific operator disable is active".to_string(),
697 });
698 }
699
700 let chain = operator.chain(pair.chain_id).ok_or_else(|| {
701 PriceOracleError::InvalidConfiguration(format!(
702 "{} references unknown operator chain_id {}",
703 pair.pair(),
704 pair.chain_id
705 ))
706 })?;
707 if !chain.enabled {
708 return Err(PriceOracleError::ChainDisabled {
709 pair: pair.pair(),
710 chain_id: chain.chain_id,
711 });
712 }
713
714 if let Some(status) =
715 read_sequencer_status(chain, now, &self.config.egress_contract).await?
716 {
717 match status.availability {
718 SequencerAvailability::Up => {}
719 SequencerAvailability::Down => {
720 return Err(PriceOracleError::SequencerDown {
721 pair: pair.pair(),
722 chain_id: chain.chain_id,
723 feed_address: status.feed_address,
724 });
725 }
726 SequencerAvailability::Recovering => {
727 let elapsed = now.saturating_sub(status.status_started_at);
728 return Err(PriceOracleError::SequencerRecovering {
729 pair: pair.pair(),
730 chain_id: chain.chain_id,
731 feed_address: status.feed_address,
732 remaining_seconds: chain
733 .sequencer_grace_period_seconds
734 .saturating_sub(elapsed),
735 });
736 }
737 }
738 }
739
740 Ok(())
741 }
742
743 fn pair_config(&self, base: &str, quote: &str) -> Result<Option<PairConfig>, PriceOracleError> {
744 self.config.validate()?;
745 Ok(self.config.pair(base, quote).cloned())
746 }
747
748 fn backend_for_kind(&self, kind: OracleBackendKind) -> Option<&Arc<dyn OracleBackend>> {
749 if self.primary.kind() == kind {
750 Some(&self.primary)
751 } else if self
752 .fallback
753 .as_ref()
754 .is_some_and(|backend| backend.kind() == kind)
755 {
756 self.fallback.as_ref()
757 } else {
758 None
759 }
760 }
761
762 fn effective_pair_override(
763 &self,
764 operator: &OperatorConfig,
765 pair: &PairConfig,
766 ) -> PairRuntimeOverride {
767 operator
768 .pair_override(&pair.base, &pair.quote)
769 .cloned()
770 .unwrap_or_else(|| PairRuntimeOverride::from_pair(pair))
771 }
772
773 fn secondary_backend_for_pair(
774 &self,
775 pair: &PairConfig,
776 pair_override: &PairRuntimeOverride,
777 ) -> Option<&Arc<dyn OracleBackend>> {
778 if !pair_override.allow_fallback {
779 return None;
780 }
781 let backend = self.fallback.as_ref()?;
782 if pair_supports_backend(pair, backend.kind()) {
783 Some(backend)
784 } else {
785 None
786 }
787 }
788}
789
790impl PriceOracle for ChioLinkOracle {
791 fn get_rate<'a>(&'a self, base: &'a str, quote: &'a str) -> OracleFuture<'a> {
792 Box::pin(async move {
793 let pair = self.pair_config(base, quote)?.ok_or_else(|| {
794 PriceOracleError::NoPairAvailable {
795 base: base.to_ascii_uppercase(),
796 quote: quote.to_ascii_uppercase(),
797 }
798 })?;
799 let now = now_unix()?;
800 let operator = self.operator_config().await;
801 self.rate_for_pair(&pair, now, &operator).await
802 })
803 }
804
805 fn supported_pairs(&self) -> Vec<String> {
806 self.config.supported_pairs()
807 }
808}
809
810fn build_backend(
811 kind: OracleBackendKind,
812 config: &PriceOracleConfig,
813) -> Result<Arc<dyn OracleBackend>, PriceOracleError> {
814 let backend: Arc<dyn OracleBackend> = match kind {
815 OracleBackendKind::Chainlink => {
816 #[cfg(feature = "web3")]
817 {
818 Arc::new(ChainlinkFeedReader::new(
819 config.operator.chains.clone(),
820 config.egress_contract.clone(),
821 ))
822 }
823 #[cfg(not(feature = "web3"))]
824 {
825 let _ = config;
826 return Err(PriceOracleError::UnsupportedBackend(
827 "Chainlink backend requires the `web3` feature".to_string(),
828 ));
829 }
830 }
831 OracleBackendKind::Pyth => Arc::new(PythHermesClient::new(
832 config.pyth.hermes_url.clone(),
833 config.egress_contract.clone(),
834 )?),
835 };
836 Ok(backend)
837}
838
839fn pair_supports_backend(pair: &PairConfig, kind: OracleBackendKind) -> bool {
840 match kind {
841 OracleBackendKind::Chainlink => pair.chainlink.is_some(),
842 OracleBackendKind::Pyth => pair.pyth.is_some(),
843 }
844}
845
846async fn read_validated_backend_rate(
847 backend: &dyn OracleBackend,
848 pair: &PairConfig,
849 now: u64,
850) -> Result<ExchangeRate, PriceOracleError> {
851 let rate = backend.read_rate(pair, now).await?;
852 rate.ensure_matches_pair(pair)?;
853 rate.ensure_fresh(now)?;
854 Ok(rate)
855}
856
857fn degraded_rate_if_allowed(
858 pair: &PairConfig,
859 pair_override: &PairRuntimeOverride,
860 stale_rate: ExchangeRate,
861 now: u64,
862) -> Option<ExchangeRate> {
863 let degraded_mode = pair_override
864 .degraded_mode
865 .clone()
866 .unwrap_or_else(|| pair.policy.degraded_mode.clone());
867 if !degraded_mode.enabled {
868 return None;
869 }
870 let allowed_age = pair
871 .policy
872 .max_age_seconds
873 .saturating_add(degraded_mode.max_stale_age_seconds);
874 if stale_rate.age_seconds(now) > allowed_age {
875 return None;
876 }
877 Some(stale_rate.with_degraded_mode(allowed_age, degraded_mode.extra_margin_bps, "degraded"))
878}
879
880fn now_unix() -> Result<u64, PriceOracleError> {
881 SystemTime::now()
882 .duration_since(UNIX_EPOCH)
883 .map(|duration| duration.as_secs())
884 .map_err(|err| {
885 PriceOracleError::Unavailable(format!("system clock is before UNIX epoch: {err}"))
886 })
887}
888
889#[cfg(test)]
890mod tests;