evm_oracle_state/pending/
adapter.rs

1//! Oracle-family adapters for speculative transport candidates.
2
3use std::collections::BTreeSet;
4
5use alloy_primitives::{Address, keccak256};
6use alloy_rpc_types_eth::Log;
7
8use crate::{FeedRegistration, FeedSource, OracleAdapterId, decode_ocr2_new_transmission};
9
10use super::{
11    CHAINLINK_FORWARD_SELECTOR, CHAINLINK_TRANSMIT_SECONDARY_SELECTOR, CHAINLINK_TRANSMIT_SELECTOR,
12    PendingOracleEvidence, PendingOracleObserveOutcome, PendingOracleOrderingHandle,
13    PendingOracleSimulationMaterial, PendingOracleTransmission, PendingOracleUpdate,
14    PendingOracleValue, PendingTransportCandidate, decode_chainlink_ocr2_calldata,
15};
16
17/// Coarse upstream filter requested by one pending oracle adapter.
18///
19/// An empty address set means any destination. Sources may use these interests
20/// to reduce traffic, but adapters still validate every submitted candidate.
21#[derive(Clone, Debug, PartialEq, Eq)]
22pub struct PendingOracleInterest {
23    /// Adapter requesting this interest.
24    pub adapter_id: OracleAdapterId,
25    /// Eligible outer transaction destinations, or empty for any destination.
26    pub addresses: BTreeSet<Address>,
27    /// Eligible four-byte calldata selectors.
28    pub selectors: BTreeSet<[u8; 4]>,
29}
30
31impl PendingOracleInterest {
32    /// Construct an interest from destination and selector iterators.
33    pub fn new(
34        adapter_id: OracleAdapterId,
35        addresses: impl IntoIterator<Item = Address>,
36        selectors: impl IntoIterator<Item = [u8; 4]>,
37    ) -> Self {
38        Self {
39            adapter_id,
40            addresses: addresses.into_iter().collect(),
41            selectors: selectors.into_iter().collect(),
42        }
43    }
44
45    /// Return whether this interest accepts the candidate's outer call.
46    pub fn matches(&self, candidate: &PendingTransportCandidate) -> bool {
47        let Some(selector) = candidate.calldata.get(..4) else {
48            return false;
49        };
50        (self.addresses.is_empty() || self.addresses.contains(&candidate.to))
51            && self
52                .selectors
53                .iter()
54                .any(|expected| selector == expected.as_slice())
55    }
56}
57
58/// Adapter-owned pending decode failure.
59#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
60pub enum PendingOracleAdapterError {
61    /// The candidate came from a different chain than the configured runtime.
62    #[error("pending candidate is for chain {observed}, expected {expected}")]
63    WrongChain {
64        /// Configured chain id.
65        expected: u64,
66        /// Candidate chain id.
67        observed: u64,
68    },
69    /// The candidate's source family was not enabled.
70    #[error("pending source {0:?} is not enabled")]
71    SourceDisabled(super::PendingOracleSource),
72    /// An oracle-family decoder rejected otherwise matching calldata.
73    #[error("pending adapter {adapter_id} failed to decode candidate: {message}")]
74    Decode {
75        /// Adapter that rejected the candidate.
76        adapter_id: OracleAdapterId,
77        /// Stable human-readable diagnostic.
78        message: String,
79    },
80}
81
82impl PendingOracleAdapterError {
83    /// Construct an adapter decode error.
84    pub fn new(adapter_id: OracleAdapterId, message: impl Into<String>) -> Self {
85        Self::Decode {
86            adapter_id,
87            message: message.into(),
88        }
89    }
90}
91
92/// Result of recording one adapter-decoded update.
93#[derive(Clone, Debug, PartialEq, Eq)]
94pub struct PendingOracleAdapterObservation {
95    /// Adapter that decoded the update.
96    pub adapter_id: OracleAdapterId,
97    /// Stable update identity returned by that adapter.
98    pub update_id: super::PendingOracleUpdateId,
99    /// Pending tracker outcome.
100    pub outcome: PendingOracleObserveOutcome,
101}
102
103/// One adapter-local failure recorded while other adapters continued routing.
104#[non_exhaustive]
105#[derive(Clone, Debug, PartialEq, Eq)]
106pub struct PendingOracleAdapterFailure {
107    /// Adapter that failed to decode the candidate.
108    pub adapter_id: OracleAdapterId,
109    /// Adapter-provided failure.
110    pub error: PendingOracleAdapterError,
111}
112
113/// Result of routing one candidate through every interested pending adapter.
114#[non_exhaustive]
115#[derive(Clone, Debug, Default, PartialEq, Eq)]
116pub struct PendingOracleCandidateReport {
117    /// Successfully decoded and recorded observations.
118    pub observations: Vec<PendingOracleAdapterObservation>,
119    /// Adapter-local failures that did not block other adapters.
120    pub failures: Vec<PendingOracleAdapterFailure>,
121}
122
123/// Oracle-family extension point for pre-inclusion candidates.
124pub trait PendingOracleAdapter: Send + Sync + 'static {
125    /// Stable adapter identity.
126    fn adapter_id(&self) -> OracleAdapterId;
127
128    /// Describe transport-level filters for the current registered feed scope.
129    fn interests(&self, registrations: &[FeedRegistration]) -> Vec<PendingOracleInterest>;
130
131    /// Decode zero or more updates from one transport candidate.
132    fn decode(
133        &self,
134        candidate: &PendingTransportCandidate,
135        registrations: &[FeedRegistration],
136    ) -> Result<Vec<PendingOracleUpdate>, PendingOracleAdapterError>;
137
138    /// Return whether a committed log is the oracle-family confirmation for an update.
139    fn confirms(&self, _update: &PendingOracleUpdate, _log: &Log) -> bool {
140        false
141    }
142}
143
144/// Pending adapter for direct and forwarded Chainlink OCR2 reports.
145#[derive(Clone, Copy, Debug, Default)]
146pub struct ChainlinkPendingAdapter;
147
148impl ChainlinkPendingAdapter {
149    /// Stable id used by the built-in Chainlink pending adapter.
150    pub const ID: &'static str = "chainlink-ocr2";
151}
152
153impl PendingOracleAdapter for ChainlinkPendingAdapter {
154    fn adapter_id(&self) -> OracleAdapterId {
155        OracleAdapterId::new(Self::ID)
156    }
157
158    fn interests(&self, registrations: &[FeedRegistration]) -> Vec<PendingOracleInterest> {
159        let aggregators = registrations
160            .iter()
161            .filter(|registration| matches!(&registration.source, FeedSource::Chainlink))
162            .filter_map(|registration| registration.current_aggregator);
163        let aggregators = aggregators.collect::<BTreeSet<_>>();
164        if aggregators.is_empty() {
165            return Vec::new();
166        }
167        vec![
168            PendingOracleInterest::new(
169                self.adapter_id(),
170                aggregators,
171                [
172                    CHAINLINK_TRANSMIT_SELECTOR,
173                    CHAINLINK_TRANSMIT_SECONDARY_SELECTOR,
174                ],
175            ),
176            PendingOracleInterest::new(self.adapter_id(), [], [CHAINLINK_FORWARD_SELECTOR]),
177        ]
178    }
179
180    fn decode(
181        &self,
182        candidate: &PendingTransportCandidate,
183        registrations: &[FeedRegistration],
184    ) -> Result<Vec<PendingOracleUpdate>, PendingOracleAdapterError> {
185        let Some(decoded) = decode_chainlink_ocr2_calldata(candidate.to, &candidate.calldata)
186            .map_err(|error| {
187                PendingOracleAdapterError::new(self.adapter_id(), error.to_string())
188            })?
189        else {
190            return Ok(Vec::new());
191        };
192        let feed_ids = registrations
193            .iter()
194            .filter(|registration| {
195                matches!(&registration.source, FeedSource::Chainlink)
196                    && registration.current_aggregator == Some(decoded.report.aggregator)
197            })
198            .map(|registration| registration.id.clone())
199            .collect::<Vec<_>>();
200        if feed_ids.is_empty() {
201            return Ok(Vec::new());
202        }
203
204        let simulation = match &candidate.ordering {
205            PendingOracleOrderingHandle::RawEthereumTransaction {
206                tx_hash,
207                signed_envelope,
208            } if keccak256(signed_envelope) == *tx_hash => {
209                PendingOracleSimulationMaterial::ExactSignedTransaction
210            }
211            _ => PendingOracleSimulationMaterial::AnswerOnlyProjection,
212        };
213        let transmission = PendingOracleTransmission::new(
214            candidate.id,
215            candidate.source_id.clone(),
216            decoded.route.clone(),
217            candidate.ordering.clone(),
218            simulation,
219            decoded.report.calldata_hash,
220            candidate.observed_at_head,
221        );
222        Ok(vec![
223            PendingOracleUpdate::new(
224                decoded.update_id(),
225                feed_ids,
226                PendingOracleEvidence::ChainlinkReport(decoded.report.clone()),
227            )
228            .with_proposed_value(PendingOracleValue {
229                raw_answer: decoded.report.median_answer,
230                observed_at: decoded.report.observations_timestamp,
231            })
232            .with_transmission(transmission),
233        ])
234    }
235
236    fn confirms(&self, update: &PendingOracleUpdate, log: &Log) -> bool {
237        let PendingOracleEvidence::ChainlinkReport(report) = &update.evidence else {
238            return false;
239        };
240        decode_ocr2_new_transmission(log).is_ok_and(|confirmed| {
241            !confirmed.removed
242                && confirmed.aggregator == report.aggregator
243                && confirmed.answer == report.median_answer
244                && confirmed.observations_timestamp == report.observations_timestamp
245        })
246    }
247}
248
249pub(crate) fn adapter_owns_update(
250    adapter_id: &OracleAdapterId,
251    update: &PendingOracleUpdate,
252) -> bool {
253    match &update.evidence {
254        PendingOracleEvidence::ChainlinkReport(_) => {
255            adapter_id.as_str() == ChainlinkPendingAdapter::ID
256        }
257        PendingOracleEvidence::Adapter {
258            adapter_id: owner, ..
259        } => owner == adapter_id,
260    }
261}
evm_oracle_state/pending/adapter.rs

evm_oracle_state/pending/
adapter.rs
