evm_oracle_state/
pyth.rs

1use std::{
2    borrow::Cow,
3    collections::{BTreeMap, BTreeSet},
4    sync::Arc,
5};
6
7use alloy_network::Ethereum;
8use alloy_primitives::{Address, B256, I256, U256, keccak256};
9use alloy_rpc_types_eth::Filter;
10use alloy_sol_types::sol;
11use evm_fork_cache::{
12    StateView,
13    cache::EvmCache,
14    reactive::{
15        ChainStatus, HandlerError, HandlerId, HandlerOutcome, HookSignal, InvalidationReason,
16        InvalidationRequest, LogInterest, ReactiveContext, ReactiveEffect, ReactiveHandler,
17        ReactiveInput, ReactiveInterest, ReportTag, RouteKeySpec, StateEffectQuality,
18    },
19    state_update::PurgeScope,
20};
21
22use crate::{
23    AdapterFuture, FeedId, FeedMetadata, FeedRegistration, FeedSource, ORACLE_SIGNAL_NAMESPACE,
24    OracleAdapterId, OracleAdapterPlugin, OracleDiscoveredFeed, OracleDiscoveryContext,
25    OracleDiscoveryReport, OracleError, OracleFeedStatus, OraclePriceUpdate, OracleSignalKind,
26    OracleStorageSync, OracleValueSource, OracleValueStatus, PYTH_PRICE_FEED_UPDATE_TOPIC,
27    RoundData, StalenessPolicy, decode_pyth_price_feed_update, state::classify_round,
28};
29
30const ADAPTER_ID: &str = "evm-oracle-state.pyth";
31const HANDLER_ID: &str = "evm-oracle-state.pyth";
32
33sol! {
34    struct PythPrice {
35        int64 price;
36        uint64 conf;
37        int32 expo;
38        uint256 publishTime;
39    }
40
41    interface PythInterface {
42        function getPriceUnsafe(bytes32 id) external view returns (PythPrice memory price);
43    }
44}
45
46use PythInterface::getPriceUnsafeCall;
47
48/// Builder-style Pyth EVM feed registration config.
49#[derive(Clone, Debug, PartialEq, Eq)]
50pub struct PythFeed {
51    pyth: Address,
52    price_id: B256,
53    state_key: Option<Address>,
54    id: Option<FeedId>,
55    label: Option<String>,
56    base: Option<String>,
57    quote: Option<String>,
58    staleness: StalenessPolicy,
59}
60
61impl PythFeed {
62    /// Create a Pyth feed config from the shared Pyth contract and price id.
63    pub fn new(pyth: Address, price_id: B256) -> Self {
64        Self {
65            pyth,
66            price_id,
67            state_key: None,
68            id: None,
69            label: None,
70            base: None,
71            quote: None,
72            staleness: StalenessPolicy::max_age(300),
73        }
74    }
75
76    /// Set the typed feed id.
77    pub fn id(mut self, id: impl Into<String>) -> Self {
78        self.id = Some(FeedId::new(id));
79        self
80    }
81
82    /// Set the human-readable label.
83    pub fn label(mut self, label: impl Into<String>) -> Self {
84        self.label = Some(label.into());
85        self
86    }
87
88    /// Set the base symbol.
89    pub fn base(mut self, base: impl Into<String>) -> Self {
90        self.base = Some(base.into());
91        self
92    }
93
94    /// Set the quote symbol.
95    pub fn quote(mut self, quote: impl Into<String>) -> Self {
96        self.quote = Some(quote.into());
97        self
98    }
99
100    /// Set the max-age staleness policy.
101    pub fn max_age_secs(mut self, max_age_secs: u64) -> Self {
102        self.staleness = StalenessPolicy::max_age(max_age_secs);
103        self
104    }
105
106    /// Set an explicit per-feed state key.
107    pub fn state_key(mut self, state_key: Address) -> Self {
108        self.state_key = Some(state_key);
109        self
110    }
111
112    /// Shared Pyth contract address.
113    pub fn pyth(&self) -> Address {
114        self.pyth
115    }
116
117    /// Pyth price feed id.
118    pub fn price_id(&self) -> B256 {
119        self.price_id
120    }
121
122    /// Synthetic per-feed state key used as the registry proxy.
123    pub fn proxy(&self) -> Address {
124        self.state_key
125            .unwrap_or_else(|| Self::state_key_for(self.pyth, self.price_id))
126    }
127
128    /// Deterministically derive a synthetic per-feed state key.
129    pub fn state_key_for(pyth: Address, price_id: B256) -> Address {
130        let mut bytes = Vec::with_capacity(32 + 20 + 32);
131        bytes.extend_from_slice(ADAPTER_ID.as_bytes());
132        bytes.extend_from_slice(pyth.as_slice());
133        bytes.extend_from_slice(price_id.as_slice());
134        let hash = keccak256(bytes);
135        Address::from_slice(&hash.as_slice()[12..])
136    }
137
138    fn feed_id(&self) -> FeedId {
139        self.id.clone().unwrap_or_else(|| {
140            let price_id = self.price_id;
141            FeedId::new(format!("pyth-{price_id:?}"))
142        })
143    }
144
145    fn description(&self) -> String {
146        self.label
147            .clone()
148            .unwrap_or_else(|| match (&self.base, &self.quote) {
149                (Some(base), Some(quote)) => format!("Pyth {base}/{quote}"),
150                _ => {
151                    let price_id = self.price_id;
152                    format!("Pyth {price_id:?}")
153                }
154            })
155    }
156}
157
158/// Pyth EVM oracle adapter plugin.
159#[derive(Clone, Debug, Default)]
160pub struct PythOracleAdapter {
161    feeds: Vec<PythFeed>,
162}
163
164impl PythOracleAdapter {
165    /// Create an empty Pyth adapter.
166    pub fn new() -> Self {
167        Self::default()
168    }
169
170    /// Add one Pyth feed config.
171    pub fn feed(mut self, feed: PythFeed) -> Self {
172        self.feeds.push(feed);
173        self
174    }
175
176    /// Add multiple Pyth feed configs.
177    pub fn feeds(mut self, feeds: impl IntoIterator<Item = PythFeed>) -> Self {
178        self.feeds.extend(feeds);
179        self
180    }
181}
182
183impl OracleAdapterPlugin for PythOracleAdapter {
184    fn adapter_id(&self) -> OracleAdapterId {
185        OracleAdapterId::new(ADAPTER_ID)
186    }
187
188    fn discover<'a>(
189        &'a self,
190        ctx: OracleDiscoveryContext<'a>,
191    ) -> AdapterFuture<'a, OracleDiscoveryReport> {
192        Box::pin(async move {
193            let mut report = OracleDiscoveryReport::new();
194            for feed in &self.feeds {
195                let price = read_pyth_price(ctx.cache, feed.pyth, feed.price_id)?;
196                let (answer, decimals) = scale_pyth_price(price.price, price.expo)?;
197                let publish_time = u256_to_u64(price.publishTime, "publishTime")?;
198                let id = feed.feed_id();
199                let metadata = FeedMetadata {
200                    decimals,
201                    description: feed.description(),
202                    version: U256::ZERO,
203                };
204                let registration = FeedRegistration {
205                    id,
206                    proxy: feed.proxy(),
207                    label: feed.label.clone(),
208                    base: feed.base.clone(),
209                    quote: feed.quote.clone(),
210                    staleness: feed.staleness,
211                    current_aggregator: Some(feed.pyth),
212                    aggregator_layout: None,
213                    metadata,
214                    source: FeedSource::pyth(feed.pyth, feed.price_id, price.expo, price.conf),
215                    status: OracleFeedStatus::Ready,
216                };
217                let round = RoundData {
218                    round_id: U256::from(publish_time),
219                    answer,
220                    started_at: publish_time,
221                    updated_at: publish_time,
222                    answered_in_round: U256::from(publish_time),
223                };
224                report = report.with_feed(OracleDiscoveredFeed::new(registration, round));
225            }
226            Ok(report)
227        })
228    }
229
230    fn reactive_handler(
231        &self,
232        registrations: Vec<FeedRegistration>,
233        _storage_sync: OracleStorageSync,
234    ) -> Arc<dyn ReactiveHandler<Ethereum>> {
235        Arc::new(PythReactiveHandler::new(registrations))
236    }
237}
238
239/// Reactive handler for Pyth EVM `PriceFeedUpdate` logs.
240#[derive(Clone, Debug)]
241pub struct PythReactiveHandler {
242    registrations_by_price: BTreeMap<(Address, B256), Vec<FeedRegistration>>,
243    pyth_contracts: BTreeSet<Address>,
244}
245
246impl PythReactiveHandler {
247    /// Create a Pyth handler from current feed registrations.
248    pub fn new(registrations: Vec<FeedRegistration>) -> Self {
249        let mut registrations_by_price: BTreeMap<(Address, B256), Vec<FeedRegistration>> =
250            BTreeMap::new();
251        let mut pyth_contracts = BTreeSet::new();
252
253        for registration in registrations {
254            let Some((pyth, price_id, _expo, _conf)) = registration.source.pyth_source() else {
255                continue;
256            };
257            pyth_contracts.insert(pyth);
258            registrations_by_price
259                .entry((pyth, price_id))
260                .or_default()
261                .push(registration);
262        }
263
264        Self {
265            registrations_by_price,
266            pyth_contracts,
267        }
268    }
269
270    /// Stable handler id.
271    pub fn id(&self) -> HandlerId {
272        HandlerId::new(HANDLER_ID)
273    }
274
275    /// Interests for registered Pyth contracts.
276    pub fn interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
277        self.pyth_contracts
278            .iter()
279            .map(|pyth| {
280                ReactiveInterest::Logs(LogInterest {
281                    provider_filter: Filter::new()
282                        .address(*pyth)
283                        .event_signature(PYTH_PRICE_FEED_UPDATE_TOPIC),
284                    local_matcher: None,
285                    route_key: Some(RouteKeySpec::EmitterAddress),
286                })
287            })
288            .collect()
289    }
290}
291
292impl ReactiveHandler<Ethereum> for PythReactiveHandler {
293    fn id(&self) -> HandlerId {
294        self.id()
295    }
296
297    fn interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
298        self.interests()
299    }
300
301    fn handle(
302        &self,
303        ctx: &ReactiveContext,
304        input: &ReactiveInput<Ethereum>,
305        _state: &dyn StateView,
306    ) -> Result<HandlerOutcome, HandlerError> {
307        let ReactiveInput::Log(log) = input else {
308            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
309        };
310        if log.removed && !matches!(ctx.chain_status, ChainStatus::Reorged { .. }) {
311            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
312        }
313        if log.topics().first().copied() != Some(PYTH_PRICE_FEED_UPDATE_TOPIC) {
314            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
315        }
316
317        let event = decode_pyth_price_feed_update(log).map_err(|error| {
318            HandlerError::new(format!("decode Pyth PriceFeedUpdate failed: {error}"))
319        })?;
320        let key = (event.pyth, event.price_id);
321        let Some(registrations) = self.registrations_by_price.get(&key) else {
322            return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
323        };
324
325        let block_number = event
326            .block_number
327            .or_else(|| ctx.block.as_ref().map(|block| block.number));
328        let block_hash = log
329            .block_hash
330            .or_else(|| ctx.block.as_ref().map(|block| block.hash));
331        let log_index = event.log_index.or(ctx.log_index);
332        let value_status = if log.removed {
333            OracleValueStatus::RequiresRepair
334        } else {
335            OracleValueStatus::EventPending
336        };
337
338        let mut effects = vec![ReactiveEffect::Invalidate(InvalidationRequest {
339            scope: PurgeScope::AllStorage,
340            address: event.pyth,
341            reason: InvalidationReason::HandlerRequested,
342        })];
343        let mut tags = Vec::new();
344
345        for registration in registrations {
346            let (_pyth, _price_id, expo, _conf) =
347                registration.source.pyth_source().ok_or_else(|| {
348                    HandlerError::new("matched Pyth registration missing Pyth source")
349                })?;
350            let (answer, decimals) = scale_pyth_price(event.price, expo)
351                .map_err(|error| HandlerError::new(format!("{error}")))?;
352            let round = RoundData {
353                round_id: U256::from(event.publish_time),
354                answer,
355                started_at: event.publish_time,
356                updated_at: event.publish_time,
357                answered_in_round: U256::from(event.publish_time),
358            };
359            let round_status = classify_round(&round, event.publish_time, &registration.staleness);
360            let hook_tags =
361                pyth_hook_tags(registration, event.pyth, event.price_id, expo, event.conf);
362            tags.extend(hook_tags.clone());
363            effects.push(ReactiveEffect::Hook(HookSignal {
364                namespace: Cow::Borrowed(ORACLE_SIGNAL_NAMESPACE),
365                kind: Cow::Borrowed(OracleSignalKind::PriceUpdate.as_str()),
366                labels: hook_tags,
367                payload: Some(Arc::new(OraclePriceUpdate {
368                    id: registration.id.clone(),
369                    proxy: registration.proxy,
370                    aggregator: event.pyth,
371                    label: registration.label.clone(),
372                    base: registration.base.clone(),
373                    quote: registration.quote.clone(),
374                    raw_answer: answer,
375                    decimals,
376                    event_round_id: U256::from(event.publish_time),
377                    started_at: event.publish_time,
378                    updated_at: event.publish_time,
379                    block_number,
380                    block_hash,
381                    log_index,
382                    round_status,
383                    value_status,
384                    source: OracleValueSource::Event,
385                })),
386            }));
387        }
388
389        Ok(HandlerOutcome {
390            effects,
391            quality: StateEffectQuality::RequiresRepair,
392            tags,
393        })
394    }
395}
396
397fn read_pyth_price(
398    cache: &mut EvmCache,
399    pyth: Address,
400    price_id: B256,
401) -> Result<PythPrice, OracleError> {
402    cache
403        .call_sol(pyth, getPriceUnsafeCall { id: price_id })
404        .map_err(|error| OracleError::Provider(format!("Pyth getPriceUnsafe failed: {error}")))
405}
406
407pub(crate) fn scale_pyth_price(price: i64, expo: i32) -> Result<(I256, u8), OracleError> {
408    let mut answer = I256::unchecked_from(price);
409    if expo <= 0 {
410        let decimals = expo
411            .checked_neg()
412            .and_then(|value| u8::try_from(value).ok())
413            .ok_or_else(|| OracleError::Unsupported(format!("Pyth expo {expo} is too negative")))?;
414        return Ok((answer, decimals));
415    }
416
417    let scale = I256::unchecked_from(10_i8)
418        .checked_pow(U256::from(expo as u32))
419        .ok_or_else(|| OracleError::Unsupported(format!("Pyth expo {expo} scale overflowed")))?;
420    answer = answer
421        .checked_mul(scale)
422        .ok_or_else(|| OracleError::Unsupported("Pyth scaled price overflowed".to_string()))?;
423    Ok((answer, 0))
424}
425
426fn pyth_hook_tags(
427    registration: &FeedRegistration,
428    pyth: Address,
429    price_id: B256,
430    expo: i32,
431    conf: u64,
432) -> Vec<ReportTag> {
433    vec![
434        ReportTag::new("feed_id", registration.id.to_string()),
435        ReportTag::new("proxy", format!("{:?}", registration.proxy)),
436        ReportTag::new("aggregator", format!("{pyth:?}")),
437        ReportTag::new("pyth_price_id", format!("{price_id:?}")),
438        ReportTag::new("pyth_expo", expo.to_string()),
439        ReportTag::new("pyth_conf", conf.to_string()),
440    ]
441}
442
443fn u256_to_u64(value: U256, field: &'static str) -> Result<u64, OracleError> {
444    u64::try_from(value).map_err(|_| {
445        OracleError::Decode(crate::ChainlinkEventDecodeError::Uint64Overflow { field, value })
446    })
447}
evm_oracle_state/pyth.rs

evm_oracle_state/
pyth.rs
