1use std::{
2 borrow::Cow,
3 collections::{BTreeMap, BTreeSet},
4 sync::Arc,
5 time::{SystemTime, UNIX_EPOCH},
6};
7
8use alloy_network::Ethereum;
9use alloy_primitives::{Address, B256, I256, U256, keccak256};
10use alloy_rpc_types_eth::Filter;
11use alloy_sol_types::sol;
12use evm_fork_cache::{
13 StateUpdate, StateView,
14 cache::EvmCache,
15 reactive::{
16 ChainStatus, HandlerError, HandlerId, HandlerOutcome, HookSignal, InvalidationReason,
17 InvalidationRequest, LogInterest, ReactiveContext, ReactiveEffect, ReactiveHandler,
18 ReactiveInput, ReactiveInterest, ReportTag, RouteKeySpec, StateEffectQuality,
19 },
20 state_update::PurgeScope,
21};
22
23use crate::{
24 ANSWER_UPDATED_TOPIC, AdapterFuture, AssetId, Denomination, EvmCacheChainlinkReader, Feed,
25 FeedConfig, FeedId, FeedMetadata, FeedRegistration, FeedSource, ORACLE_SIGNAL_NAMESPACE,
26 OracleAdapterFeedSkip, OracleAdapterId, OracleAdapterPlugin, OracleAdapterSkipReason,
27 OracleDiscoveredFeed, OracleDiscoveryContext, OracleDiscoveryReport, OracleError,
28 OracleFeedStatus, OraclePriceUpdate, OracleSignalKind, OracleStorageSync, OracleValueSource,
29 OracleValueStatus, REDSTONE_VALUE_UPDATE_TOPIC, RedstoneValueUpdate, RoundData,
30 StalenessPolicy, decode_answer_updated, decode_redstone_value_update, state::classify_round,
31};
32
33sol! {
34 interface RedstonePriceFeedInterface {
35 function getDataFeedId() external view returns (bytes32);
36 function getPriceFeedAdapter() external view returns (address);
37 }
38}
39
40use RedstonePriceFeedInterface::{getDataFeedIdCall, getPriceFeedAdapterCall};
41
42const ADAPTER_ID: &str = "evm-oracle-state.redstone";
43const HANDLER_ID: &str = "evm-oracle-state.redstone";
44const REDSTONE_NO_ROUNDS_ROUND_ID: u64 = 1;
45const REDSTONE_MULTI_FEED_DATA_FEEDS_STORAGE_LOCATION: B256 =
46 alloy_primitives::b256!("5e9fb4cb0eb3c2583734d3394f30bb14b241acb9b3a034f7e7ba1a62db4370f1");
47const REDSTONE_PRICE_FEEDS_VALUES_MAPPING_STORAGE_LOCATION: B256 =
48 alloy_primitives::b256!("4dd0c77efa6f6d590c97573d8c70b714546e7311202ff7c11c484cc841d91bfc");
49const REDSTONE_PRICE_FEEDS_LATEST_UPDATE_TIMESTAMPS_STORAGE_LOCATION: B256 =
50 alloy_primitives::b256!("3d01e4d77237ea0f771f1786da4d4ff757fcba6a92933aa53b1dcef2d6bd6fe2");
51const REDSTONE_PRICE_FEEDS_WITH_ROUNDS_ROUND_TIMESTAMPS_MAPPING_STORAGE_LOCATION: B256 =
52 alloy_primitives::b256!("207e00944d909d1224f0c253d58489121d736649f8393199f55eecf4f0cf3eb0");
53const REDSTONE_PRICE_FEEDS_WITH_ROUNDS_LATEST_ROUND_ID_STORAGE_LOCATION: B256 =
54 alloy_primitives::b256!("c68d7f1ee07d8668991a8951e720010c9d44c2f11c06b5cac61fbc4083263938");
55const REDSTONE_DATA_TIMESTAMP_BITS: usize = 48;
56const REDSTONE_BLOCK_TIMESTAMP_BITS: usize = 48;
57const REDSTONE_MULTI_FEED_VALUE_BITS: usize = 152;
58const REDSTONE_MULTI_FEED_BLOCK_TIMESTAMP_OFFSET_BITS: usize = REDSTONE_DATA_TIMESTAMP_BITS;
59const REDSTONE_MULTI_FEED_VALUE_OFFSET_BITS: usize =
60 REDSTONE_DATA_TIMESTAMP_BITS + REDSTONE_BLOCK_TIMESTAMP_BITS;
61const REDSTONE_MULTI_FEED_IS_VALUE_BIGGER_OFFSET_BITS: usize =
62 REDSTONE_MULTI_FEED_VALUE_OFFSET_BITS + REDSTONE_MULTI_FEED_VALUE_BITS;
63const REDSTONE_PRICE_FEEDS_BLOCK_TIMESTAMP_OFFSET_BITS: usize = 128;
64
65#[derive(Clone, Debug, PartialEq, Eq)]
67pub struct RedstoneFeed {
68 price_feed: Address,
69 adapter: Option<Address>,
70 data_feed_id: Option<B256>,
71 feed_id: Option<FeedId>,
72 label: Option<String>,
73 base: Option<AssetId>,
74 quote: Option<Denomination>,
75 staleness: StalenessPolicy,
76}
77
78impl RedstoneFeed {
79 pub fn new(price_feed: Address) -> Self {
81 Self {
82 price_feed,
83 adapter: None,
84 data_feed_id: None,
85 feed_id: None,
86 label: None,
87 base: None,
88 quote: None,
89 staleness: StalenessPolicy::default(),
90 }
91 }
92
93 pub fn price_feed(price_feed: Address) -> Self {
95 Self::new(price_feed)
96 }
97
98 pub fn push(price_feed: Address, adapter: Address, data_feed_id: B256) -> Self {
100 Self::new(price_feed)
101 .adapter(adapter)
102 .data_feed_id(data_feed_id)
103 }
104
105 pub fn price_feed_address(&self) -> Address {
107 self.price_feed
108 }
109
110 pub fn adapter(mut self, adapter: Address) -> Self {
112 self.adapter = Some(adapter);
113 self
114 }
115
116 pub fn data_feed_id(mut self, data_feed_id: B256) -> Self {
118 self.data_feed_id = Some(data_feed_id);
119 self
120 }
121
122 pub fn id(mut self, id: impl Into<String>) -> Self {
124 self.feed_id = Some(FeedId::new(id));
125 self
126 }
127
128 pub fn feed_id(mut self, id: FeedId) -> Self {
130 self.feed_id = Some(id);
131 self
132 }
133
134 pub fn label(mut self, label: impl Into<String>) -> Self {
136 self.label = Some(label.into());
137 self
138 }
139
140 pub fn base(mut self, base: AssetId) -> Self {
142 self.base = Some(base);
143 self
144 }
145
146 pub fn quote(mut self, quote: Denomination) -> Self {
148 self.quote = Some(quote);
149 self
150 }
151
152 pub fn max_age_secs(mut self, max_age_secs: u64) -> Self {
154 self.staleness = StalenessPolicy::max_age(max_age_secs);
155 self
156 }
157
158 pub fn staleness(mut self, staleness: StalenessPolicy) -> Self {
160 self.staleness = staleness;
161 self
162 }
163
164 fn feed_for_skip(&self) -> Feed {
165 let mut feed = Feed::proxy(self.price_feed);
166 if let Some(id) = self.feed_id.clone() {
167 feed = feed.feed_id(id);
168 }
169 if let Some(label) = &self.label {
170 feed = feed.label(label.clone());
171 }
172 if let Some(base) = &self.base {
173 feed = feed.base(base.clone());
174 }
175 if let Some(quote) = &self.quote {
176 feed = feed.quote(quote.clone());
177 }
178 feed.staleness(self.staleness)
179 }
180
181 fn config(&self) -> FeedConfig {
182 FeedConfig {
183 proxy: self.price_feed,
184 id: self.feed_id.clone(),
185 label: self.label.clone(),
186 base: self.base.clone().map(String::from),
187 quote: self.quote.clone().map(String::from),
188 staleness: self.staleness,
189 }
190 }
191}
192
193#[derive(Clone, Debug, Default)]
195pub struct RedstoneOracleAdapter {
196 feeds: Vec<RedstoneFeed>,
197 now_timestamp: Option<u64>,
198}
199
200impl RedstoneOracleAdapter {
201 pub fn new() -> Self {
203 Self::default()
204 }
205
206 pub fn price_feed(price_feed: Address) -> Self {
208 Self::new().feed(RedstoneFeed::new(price_feed))
209 }
210
211 pub fn feed(mut self, feed: RedstoneFeed) -> Self {
213 self.feeds.push(feed);
214 self
215 }
216
217 pub fn feeds(mut self, feeds: impl IntoIterator<Item = RedstoneFeed>) -> Self {
219 self.feeds.extend(feeds);
220 self
221 }
222
223 pub fn now_timestamp(mut self, now_timestamp: u64) -> Self {
225 self.now_timestamp = Some(now_timestamp);
226 self
227 }
228
229 fn timestamp(&self, fallback: Option<u64>) -> Result<u64, OracleError> {
230 if let Some(now_timestamp) = self.now_timestamp.or(fallback) {
231 return Ok(now_timestamp);
232 }
233 Ok(SystemTime::now()
234 .duration_since(UNIX_EPOCH)
235 .map_err(crate::error::clock_error)?
236 .as_secs())
237 }
238
239 fn discover_feeds(
240 &self,
241 cache: &mut EvmCache,
242 _now_timestamp: u64,
243 ) -> Result<OracleDiscoveryReport, OracleError> {
244 let mut report = OracleDiscoveryReport::new();
245 for feed in &self.feeds {
246 match self.discover_feed(cache, feed) {
247 Ok(discovered) => report = report.with_feed(discovered),
248 Err(error) => {
249 report = report.with_skip(OracleAdapterFeedSkip {
250 feed: feed.feed_for_skip(),
251 proxy: feed.price_feed,
252 reason: OracleAdapterSkipReason::UnsupportedRedstoneSource {
253 error: error.to_string(),
254 },
255 });
256 }
257 }
258 }
259 Ok(report)
260 }
261
262 fn discover_feed(
263 &self,
264 cache: &mut EvmCache,
265 feed: &RedstoneFeed,
266 ) -> Result<OracleDiscoveredFeed, OracleError> {
267 let data_feed_id = match feed.data_feed_id {
268 Some(data_feed_id) => data_feed_id,
269 None => cache
270 .call_sol(feed.price_feed, getDataFeedIdCall {})
271 .map_err(provider_error)?,
272 };
273 let adapter = match feed.adapter {
274 Some(adapter) => adapter,
275 None => cache
276 .call_sol(feed.price_feed, getPriceFeedAdapterCall {})
277 .map_err(provider_error)?,
278 };
279
280 let reader = EvmCacheChainlinkReader::new(cache);
281 let metadata = FeedMetadata {
282 decimals: reader.read_decimals(feed.price_feed)?,
283 description: reader.read_description(feed.price_feed)?,
284 version: reader.read_version(feed.price_feed)?,
285 };
286 let round = reader.read_latest_round_data(feed.price_feed)?;
287 let config = feed.config();
288 let id = config
289 .id
290 .unwrap_or_else(|| derive_redstone_feed_id(config.label.as_deref(), feed.price_feed));
291 let registration = FeedRegistration {
292 id,
293 proxy: config.proxy,
294 label: config.label,
295 base: config.base,
296 quote: config.quote,
297 staleness: config.staleness,
298 current_aggregator: Some(adapter),
299 aggregator_layout: None,
300 metadata,
301 source: FeedSource::redstone_push(feed.price_feed, adapter, data_feed_id),
302 status: OracleFeedStatus::Ready,
303 };
304
305 Ok(OracleDiscoveredFeed::new(registration, round))
306 }
307}
308
309impl OracleAdapterPlugin for RedstoneOracleAdapter {
310 fn adapter_id(&self) -> OracleAdapterId {
311 OracleAdapterId::new(ADAPTER_ID)
312 }
313
314 fn discover<'a>(
315 &'a self,
316 ctx: OracleDiscoveryContext<'a>,
317 ) -> AdapterFuture<'a, OracleDiscoveryReport> {
318 Box::pin(async move {
319 let now_timestamp = self.timestamp(Some(ctx.now_timestamp))?;
320 self.discover_feeds(ctx.cache, now_timestamp)
321 })
322 }
323
324 fn reactive_handler(
325 &self,
326 registrations: Vec<FeedRegistration>,
327 _storage_sync: OracleStorageSync,
328 ) -> Arc<dyn ReactiveHandler<Ethereum>> {
329 Arc::new(RedstoneReactiveHandler::new(registrations))
330 }
331}
332
333#[derive(Clone, Debug, Default)]
340pub struct RedstoneMultiFeedStorageAdapter;
341
342impl RedstoneMultiFeedStorageAdapter {
343 pub fn data_feed_details_slot(data_feed_id: B256) -> U256 {
345 keyed_slot(
346 data_feed_id,
347 REDSTONE_MULTI_FEED_DATA_FEEDS_STORAGE_LOCATION,
348 )
349 }
350
351 pub fn bigger_value_slot(data_feed_id: B256) -> U256 {
353 Self::data_feed_details_slot(data_feed_id) + U256::from(1_u8)
354 }
355
356 pub fn pack_data_feed_details_from_event(value: U256, updated_at: u64) -> Option<U256> {
363 let data_timestamp_ms = updated_at.checked_mul(1_000)?;
364 Self::pack_data_feed_details(value, data_timestamp_ms, updated_at)
365 }
366
367 pub fn pack_data_feed_details(
369 value: U256,
370 data_timestamp_ms: u64,
371 block_timestamp: u64,
372 ) -> Option<U256> {
373 if data_timestamp_ms > uint_mask_u64(REDSTONE_DATA_TIMESTAMP_BITS)
374 || block_timestamp > uint_mask_u64(REDSTONE_BLOCK_TIMESTAMP_BITS)
375 {
376 return None;
377 }
378
379 let inline_value = value & uint_mask(REDSTONE_MULTI_FEED_VALUE_BITS);
380 let is_value_bigger = if value > uint_mask(REDSTONE_MULTI_FEED_VALUE_BITS) {
381 U256::from(1_u8)
382 } else {
383 U256::ZERO
384 };
385
386 Some(
387 U256::from(data_timestamp_ms)
388 | (U256::from(block_timestamp) << REDSTONE_MULTI_FEED_BLOCK_TIMESTAMP_OFFSET_BITS)
389 | (inline_value << REDSTONE_MULTI_FEED_VALUE_OFFSET_BITS)
390 | (is_value_bigger << REDSTONE_MULTI_FEED_IS_VALUE_BIGGER_OFFSET_BITS),
391 )
392 }
393
394 fn state_updates_for_value_update(
395 adapter: Address,
396 event: &RedstoneValueUpdate,
397 state: &dyn StateView,
398 ) -> Option<Vec<StateUpdate>> {
399 let details_slot = Self::data_feed_details_slot(event.data_feed_id);
400 state.storage(adapter, details_slot)?;
401 let details = Self::pack_data_feed_details_from_event(event.value, event.updated_at)?;
402 let mut updates = vec![StateUpdate::slot(adapter, details_slot, details)];
403
404 if event.value > uint_mask(REDSTONE_MULTI_FEED_VALUE_BITS) {
405 let bigger_value_slot = Self::bigger_value_slot(event.data_feed_id);
406 state.storage(adapter, bigger_value_slot)?;
407 updates.push(StateUpdate::slot(adapter, bigger_value_slot, event.value));
408 }
409
410 Some(updates)
411 }
412
413 fn state_updates_for_answer(
414 adapter: Address,
415 data_feed_id: B256,
416 value: U256,
417 updated_at: u64,
418 state: &dyn StateView,
419 ) -> Option<Vec<StateUpdate>> {
420 let event = RedstoneValueUpdate {
421 adapter,
422 data_feed_id,
423 value,
424 updated_at,
425 block_number: None,
426 log_index: None,
427 removed: false,
428 };
429 Self::state_updates_for_value_update(adapter, &event, state)
430 }
431}
432
433#[derive(Clone, Debug, Default)]
438pub struct RedstonePriceFeedsStorageAdapter;
439
440impl RedstonePriceFeedsStorageAdapter {
441 pub fn value_slot(data_feed_id: B256) -> U256 {
443 keyed_slot(
444 data_feed_id,
445 REDSTONE_PRICE_FEEDS_VALUES_MAPPING_STORAGE_LOCATION,
446 )
447 }
448
449 pub fn round_value_slot(data_feed_id: B256, round_id: U256) -> U256 {
451 keyed_slot2(
452 data_feed_id,
453 round_id,
454 REDSTONE_PRICE_FEEDS_VALUES_MAPPING_STORAGE_LOCATION,
455 )
456 }
457
458 pub fn latest_update_timestamps_slot() -> U256 {
460 U256::from_be_slice(
461 REDSTONE_PRICE_FEEDS_LATEST_UPDATE_TIMESTAMPS_STORAGE_LOCATION.as_slice(),
462 )
463 }
464
465 pub fn latest_round_id_slot() -> U256 {
467 U256::from_be_slice(
468 REDSTONE_PRICE_FEEDS_WITH_ROUNDS_LATEST_ROUND_ID_STORAGE_LOCATION.as_slice(),
469 )
470 }
471
472 pub fn round_timestamp_slot(round_id: U256) -> U256 {
474 mapping_slot(
475 round_id,
476 U256::from_be_slice(
477 REDSTONE_PRICE_FEEDS_WITH_ROUNDS_ROUND_TIMESTAMPS_MAPPING_STORAGE_LOCATION
478 .as_slice(),
479 ),
480 )
481 }
482
483 pub fn pack_latest_update_timestamps_from_event(updated_at: u64) -> Option<U256> {
485 let data_timestamp_ms = updated_at.checked_mul(1_000)?;
486 Self::pack_latest_update_timestamps(data_timestamp_ms, updated_at)
487 }
488
489 pub fn pack_latest_update_timestamps(
495 data_timestamp_ms: u64,
496 block_timestamp: u64,
497 ) -> Option<U256> {
498 Some(
499 (U256::from(data_timestamp_ms) << REDSTONE_PRICE_FEEDS_BLOCK_TIMESTAMP_OFFSET_BITS)
500 | U256::from(block_timestamp),
501 )
502 }
503
504 fn state_updates_for_value_update(
505 adapter: Address,
506 event: &RedstoneValueUpdate,
507 state: &dyn StateView,
508 ) -> Option<Vec<StateUpdate>> {
509 let value_slot = Self::value_slot(event.data_feed_id);
510 let timestamp_slot = Self::latest_update_timestamps_slot();
511 state.storage(adapter, value_slot)?;
512 state.storage(adapter, timestamp_slot)?;
513 let timestamps = Self::pack_latest_update_timestamps_from_event(event.updated_at)?;
514
515 Some(vec![
516 StateUpdate::slot(adapter, value_slot, event.value),
517 StateUpdate::slot(adapter, timestamp_slot, timestamps),
518 ])
519 }
520
521 fn state_updates_for_no_rounds_answer(
522 adapter: Address,
523 data_feed_id: B256,
524 value: U256,
525 updated_at: u64,
526 state: &dyn StateView,
527 ) -> Option<Vec<StateUpdate>> {
528 let event = RedstoneValueUpdate {
529 adapter,
530 data_feed_id,
531 value,
532 updated_at,
533 block_number: None,
534 log_index: None,
535 removed: false,
536 };
537 Self::state_updates_for_value_update(adapter, &event, state)
538 }
539
540 fn state_updates_for_round_answer(
541 adapter: Address,
542 data_feed_id: B256,
543 round_id: U256,
544 value: U256,
545 updated_at: u64,
546 state: &dyn StateView,
547 ) -> Option<Vec<StateUpdate>> {
548 let value_slot = Self::round_value_slot(data_feed_id, round_id);
549 let round_timestamp_slot = Self::round_timestamp_slot(round_id);
550 let latest_round_id_slot = Self::latest_round_id_slot();
551 state.storage(adapter, value_slot)?;
552 state.storage(adapter, round_timestamp_slot)?;
553 state.storage(adapter, latest_round_id_slot)?;
554
555 Some(vec![
556 StateUpdate::slot(adapter, value_slot, value),
557 StateUpdate::slot(adapter, round_timestamp_slot, U256::from(updated_at)),
558 StateUpdate::slot(adapter, latest_round_id_slot, round_id),
559 ])
560 }
561}
562
563#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
564struct RedstoneValueUpdateKey {
565 adapter: Address,
566 data_feed_id: B256,
567}
568
569#[derive(Clone, Debug)]
571pub struct RedstoneReactiveHandler {
572 registrations_by_value_update: BTreeMap<RedstoneValueUpdateKey, Vec<FeedRegistration>>,
573 registrations_by_answer_updated: BTreeMap<Address, Vec<FeedRegistration>>,
574}
575
576impl RedstoneReactiveHandler {
577 pub fn new(registrations: Vec<FeedRegistration>) -> Self {
579 let mut registrations_by_value_update: BTreeMap<
580 RedstoneValueUpdateKey,
581 Vec<FeedRegistration>,
582 > = BTreeMap::new();
583 let mut registrations_by_answer_updated: BTreeMap<Address, Vec<FeedRegistration>> =
584 BTreeMap::new();
585 let mut seen_value_updates = BTreeSet::new();
586 let mut seen_answer_updates = BTreeSet::new();
587
588 for registration in registrations {
589 let FeedSource::RedstonePush {
590 price_feed,
591 adapter,
592 data_feed_id,
593 } = registration.source
594 else {
595 continue;
596 };
597
598 let value_key = RedstoneValueUpdateKey {
599 adapter,
600 data_feed_id,
601 };
602 if seen_value_updates.insert((registration.id.clone(), value_key)) {
603 registrations_by_value_update
604 .entry(value_key)
605 .or_default()
606 .push(registration.clone());
607 }
608
609 if seen_answer_updates.insert((registration.id.clone(), price_feed)) {
610 registrations_by_answer_updated
611 .entry(price_feed)
612 .or_default()
613 .push(registration);
614 }
615 }
616
617 Self {
618 registrations_by_value_update,
619 registrations_by_answer_updated,
620 }
621 }
622
623 pub fn id(&self) -> HandlerId {
625 HandlerId::new(HANDLER_ID)
626 }
627
628 pub fn interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
630 let mut interests = Vec::new();
631 let mut value_adapters = BTreeSet::new();
632 for key in self.registrations_by_value_update.keys() {
633 if value_adapters.insert(key.adapter) {
634 interests.push(log_interest(key.adapter, REDSTONE_VALUE_UPDATE_TOPIC));
635 }
636 }
637 interests.extend(
638 self.registrations_by_answer_updated
639 .keys()
640 .copied()
641 .map(|price_feed| log_interest(price_feed, ANSWER_UPDATED_TOPIC)),
642 );
643 interests
644 }
645
646 fn handle_value_update(
647 &self,
648 ctx: &ReactiveContext,
649 log: &alloy_rpc_types_eth::Log,
650 state: &dyn StateView,
651 ) -> Result<HandlerOutcome, HandlerError> {
652 let event = decode_redstone_value_update(log).map_err(|error| {
653 HandlerError::new(format!("decode RedStone ValueUpdate failed: {error}"))
654 })?;
655 let key = RedstoneValueUpdateKey {
656 adapter: event.adapter,
657 data_feed_id: event.data_feed_id,
658 };
659 let Some(registrations) = self.registrations_by_value_update.get(&key) else {
660 return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
661 };
662
663 let answer = redstone_value_to_i256(event.value)?;
664 Ok(self.outcome_for_registrations(
665 ctx,
666 log,
667 registrations,
668 event.adapter,
669 answer,
670 U256::from(REDSTONE_NO_ROUNDS_ROUND_ID),
671 event.updated_at,
672 event.block_number,
673 event.log_index,
674 event.removed,
675 Some(event),
676 state,
677 ))
678 }
679
680 fn handle_answer_updated(
681 &self,
682 ctx: &ReactiveContext,
683 log: &alloy_rpc_types_eth::Log,
684 state: &dyn StateView,
685 ) -> Result<HandlerOutcome, HandlerError> {
686 let price_feed = log.address();
687 let Some(registrations) = self.registrations_by_answer_updated.get(&price_feed) else {
688 return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
689 };
690 let event = decode_answer_updated(log).map_err(|error| {
691 HandlerError::new(format!("decode RedStone AnswerUpdated failed: {error}"))
692 })?;
693
694 Ok(self.outcome_for_registrations(
695 ctx,
696 log,
697 registrations,
698 price_feed,
699 event.current,
700 event.round_id,
701 event.updated_at,
702 event.block_number,
703 event.log_index,
704 event.removed,
705 None,
706 state,
707 ))
708 }
709
710 #[allow(clippy::too_many_arguments)]
711 fn outcome_for_registrations(
712 &self,
713 ctx: &ReactiveContext,
714 log: &alloy_rpc_types_eth::Log,
715 registrations: &[FeedRegistration],
716 event_source: Address,
717 raw_answer: I256,
718 event_round_id: U256,
719 updated_at: u64,
720 block_number: Option<u64>,
721 log_index: Option<u64>,
722 removed: bool,
723 value_update: Option<RedstoneValueUpdate>,
724 state: &dyn StateView,
725 ) -> HandlerOutcome {
726 let block_number = block_number.or_else(|| ctx.block.as_ref().map(|block| block.number));
727 let block_hash = log
728 .block_hash
729 .or_else(|| ctx.block.as_ref().map(|block| block.hash));
730 let log_index = log_index.or(ctx.log_index);
731 let value_status = if removed {
732 OracleValueStatus::RequiresRepair
733 } else {
734 OracleValueStatus::EventPending
735 };
736 let mut effects = Vec::new();
737 let mut tags = Vec::new();
738 let direct_updates = if removed {
739 None
740 } else {
741 registrations.first().and_then(|registration| {
742 redstone_state_updates(
743 registration,
744 event_source,
745 raw_answer,
746 event_round_id,
747 updated_at,
748 value_update.as_ref(),
749 state,
750 )
751 })
752 };
753 let has_direct_updates = direct_updates.is_some();
754 if let Some(updates) = direct_updates {
755 effects.extend(updates.into_iter().map(ReactiveEffect::StateUpdate));
756 }
757
758 for registration in registrations {
759 if !has_direct_updates {
760 append_redstone_invalidations(&mut effects, registration, event_source);
761 }
762 let hook_tags = redstone_hook_tags(registration, event_source, value_update.as_ref());
763 tags.extend(hook_tags.clone());
764 let normalized_answer = registration
765 .source
766 .normalize_answer_from_event(Some(event_source), raw_answer);
767 let round = RoundData {
768 round_id: event_round_id,
769 answer: normalized_answer,
770 started_at: updated_at,
771 updated_at,
772 answered_in_round: event_round_id,
773 };
774 let round_status = classify_round(&round, updated_at, ®istration.staleness);
775 effects.push(ReactiveEffect::Hook(HookSignal {
776 namespace: Cow::Borrowed(ORACLE_SIGNAL_NAMESPACE),
777 kind: Cow::Borrowed(OracleSignalKind::PriceUpdate.as_str()),
778 labels: hook_tags,
779 payload: Some(Arc::new(OraclePriceUpdate {
780 id: registration.id.clone(),
781 proxy: registration.proxy,
782 aggregator: event_source,
783 label: registration.label.clone(),
784 base: registration.base.clone(),
785 quote: registration.quote.clone(),
786 raw_answer: normalized_answer,
787 decimals: registration.metadata.decimals,
788 event_round_id,
789 started_at: updated_at,
790 updated_at,
791 block_number,
792 block_hash,
793 log_index,
794 round_status,
795 value_status,
796 source: OracleValueSource::Event,
797 })),
798 }));
799 }
800
801 HandlerOutcome {
802 effects,
803 quality: if has_direct_updates {
804 StateEffectQuality::ExactFromInput
805 } else {
806 StateEffectQuality::RequiresRepair
807 },
808 tags,
809 }
810 }
811}
812
813impl ReactiveHandler<Ethereum> for RedstoneReactiveHandler {
814 fn id(&self) -> HandlerId {
815 self.id()
816 }
817
818 fn interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
819 self.interests()
820 }
821
822 fn handle(
823 &self,
824 ctx: &ReactiveContext,
825 input: &ReactiveInput<Ethereum>,
826 state: &dyn StateView,
827 ) -> Result<HandlerOutcome, HandlerError> {
828 let ReactiveInput::Log(log) = input else {
829 return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
830 };
831 if log.removed && !matches!(ctx.chain_status, ChainStatus::Reorged { .. }) {
832 return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect));
833 }
834
835 match log.topics().first().copied() {
836 Some(REDSTONE_VALUE_UPDATE_TOPIC) => self.handle_value_update(ctx, log, state),
837 Some(ANSWER_UPDATED_TOPIC) => self.handle_answer_updated(ctx, log, state),
838 _ => Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect)),
839 }
840 }
841}
842
843fn log_interest(address: Address, topic: B256) -> ReactiveInterest<Ethereum> {
844 ReactiveInterest::Logs(LogInterest {
845 provider_filter: Filter::new().address(address).event_signature(topic),
846 local_matcher: None,
847 route_key: Some(RouteKeySpec::EmitterAddress),
848 })
849}
850
851fn append_redstone_invalidations(
852 effects: &mut Vec<ReactiveEffect>,
853 registration: &FeedRegistration,
854 event_source: Address,
855) {
856 let mut addresses = BTreeSet::from([registration.proxy, event_source]);
857 if let FeedSource::RedstonePush { adapter, .. } = registration.source {
858 addresses.insert(adapter);
859 }
860 effects.extend(addresses.into_iter().map(|address| {
861 ReactiveEffect::Invalidate(InvalidationRequest {
862 scope: PurgeScope::AllStorage,
863 address,
864 reason: InvalidationReason::HandlerRequested,
865 })
866 }));
867}
868
869fn redstone_hook_tags(
870 registration: &FeedRegistration,
871 event_source: Address,
872 value_update: Option<&RedstoneValueUpdate>,
873) -> Vec<ReportTag> {
874 let mut tags = vec![
875 ReportTag::new("feed_id", registration.id.to_string()),
876 ReportTag::new("proxy", format!("{:?}", registration.proxy)),
877 ReportTag::new("aggregator", format!("{event_source:?}")),
878 ];
879 if let Some(value_update) = value_update {
880 tags.push(ReportTag::new(
881 "data_feed_id",
882 format!("{:?}", value_update.data_feed_id),
883 ));
884 }
885 tags
886}
887
888fn redstone_value_to_i256(value: U256) -> Result<I256, HandlerError> {
889 I256::try_from(value)
890 .map_err(|_| HandlerError::new(format!("RedStone value {value} does not fit int256")))
891}
892
893#[allow(clippy::too_many_arguments)]
894fn redstone_state_updates(
895 registration: &FeedRegistration,
896 event_source: Address,
897 raw_answer: I256,
898 event_round_id: U256,
899 updated_at: u64,
900 value_update: Option<&RedstoneValueUpdate>,
901 state: &dyn StateView,
902) -> Option<Vec<StateUpdate>> {
903 match value_update {
904 Some(event) => redstone_state_updates_for_value_update(registration, event, state),
905 None => redstone_state_updates_for_answer_updated(
906 registration,
907 event_source,
908 raw_answer,
909 event_round_id,
910 updated_at,
911 state,
912 ),
913 }
914}
915
916fn redstone_state_updates_for_value_update(
917 registration: &FeedRegistration,
918 event: &RedstoneValueUpdate,
919 state: &dyn StateView,
920) -> Option<Vec<StateUpdate>> {
921 let FeedSource::RedstonePush {
922 adapter,
923 data_feed_id,
924 ..
925 } = registration.source
926 else {
927 return None;
928 };
929 if event.adapter != adapter || event.data_feed_id != data_feed_id {
930 return None;
931 }
932
933 RedstoneMultiFeedStorageAdapter::state_updates_for_value_update(adapter, event, state).or_else(
934 || RedstonePriceFeedsStorageAdapter::state_updates_for_value_update(adapter, event, state),
935 )
936}
937
938fn redstone_state_updates_for_answer_updated(
939 registration: &FeedRegistration,
940 event_source: Address,
941 raw_answer: I256,
942 event_round_id: U256,
943 updated_at: u64,
944 state: &dyn StateView,
945) -> Option<Vec<StateUpdate>> {
946 let FeedSource::RedstonePush {
947 price_feed,
948 adapter,
949 data_feed_id,
950 } = registration.source
951 else {
952 return None;
953 };
954 if event_source != price_feed || adapter != price_feed {
955 return None;
956 }
957 let value = redstone_i256_to_u256(raw_answer)?;
958
959 if event_round_id == U256::from(REDSTONE_NO_ROUNDS_ROUND_ID) {
960 RedstoneMultiFeedStorageAdapter::state_updates_for_answer(
961 adapter,
962 data_feed_id,
963 value,
964 updated_at,
965 state,
966 )
967 .or_else(|| {
968 RedstonePriceFeedsStorageAdapter::state_updates_for_no_rounds_answer(
969 adapter,
970 data_feed_id,
971 value,
972 updated_at,
973 state,
974 )
975 })
976 } else {
977 RedstonePriceFeedsStorageAdapter::state_updates_for_round_answer(
978 adapter,
979 data_feed_id,
980 event_round_id,
981 value,
982 updated_at,
983 state,
984 )
985 }
986}
987
988fn redstone_i256_to_u256(value: I256) -> Option<U256> {
989 let raw = value.into_raw();
990 if raw >> 255 == U256::ZERO {
991 Some(raw)
992 } else {
993 None
994 }
995}
996
997fn keyed_slot(key: B256, base_slot: B256) -> U256 {
998 let mut preimage = [0_u8; 64];
999 preimage[..32].copy_from_slice(key.as_slice());
1000 preimage[32..].copy_from_slice(base_slot.as_slice());
1001 U256::from_be_slice(keccak256(preimage).as_slice())
1002}
1003
1004fn keyed_slot2(key0: B256, key1: U256, base_slot: B256) -> U256 {
1005 let mut preimage = [0_u8; 96];
1006 preimage[..32].copy_from_slice(key0.as_slice());
1007 preimage[32..64].copy_from_slice(&key1.to_be_bytes::<32>());
1008 preimage[64..].copy_from_slice(base_slot.as_slice());
1009 U256::from_be_slice(keccak256(preimage).as_slice())
1010}
1011
1012fn mapping_slot(key: U256, base_slot: U256) -> U256 {
1013 let mut preimage = [0_u8; 64];
1014 preimage[..32].copy_from_slice(&key.to_be_bytes::<32>());
1015 preimage[32..].copy_from_slice(&base_slot.to_be_bytes::<32>());
1016 U256::from_be_slice(keccak256(preimage).as_slice())
1017}
1018
1019fn uint_mask(bits: usize) -> U256 {
1020 debug_assert!(bits <= 256);
1021 match bits {
1022 0 => U256::ZERO,
1023 256 => U256::MAX,
1024 bits => (U256::from(1_u8) << bits) - U256::from(1_u8),
1025 }
1026}
1027
1028fn uint_mask_u64(bits: usize) -> u64 {
1029 debug_assert!(bits < 64);
1032 (1_u64 << bits) - 1
1033}
1034
1035fn derive_redstone_feed_id(label: Option<&str>, price_feed: Address) -> FeedId {
1036 if let Some(label) = label {
1037 let normalized = label
1038 .chars()
1039 .filter_map(|ch| {
1040 if ch.is_ascii_alphanumeric() {
1041 Some(ch.to_ascii_lowercase())
1042 } else if ch.is_ascii_whitespace() || matches!(ch, '/' | '_' | '-') {
1043 Some('-')
1044 } else {
1045 None
1046 }
1047 })
1048 .collect::<String>()
1049 .trim_matches('-')
1050 .to_string();
1051 if !normalized.is_empty() {
1052 return FeedId::new(normalized);
1053 }
1054 }
1055 FeedId::new(format!("redstone-{price_feed:?}"))
1056}
1057
1058fn provider_error(error: impl std::fmt::Debug) -> OracleError {
1059 OracleError::Provider(format!("{error:?}"))
1060}