evm_oracle_state/
runtime.rs

1use std::{
2    collections::BTreeSet,
3    fmt,
4    sync::Arc,
5    time::{SystemTime, UNIX_EPOCH},
6};
7
8use alloy_network::{Ethereum, Network};
9use alloy_primitives::{Address, Bytes, U256};
10use alloy_sol_types::SolCall;
11use evm_fork_cache::{
12    ColdStartCall, ColdStartConfig, ColdStartPlan, ColdStartPlanner, ColdStartResults,
13    ColdStartStep, StateView,
14    cache::EvmCache,
15    reactive::{
16        EventSubscriber, HandlerId, InterestOwnerSubscriber, ReactiveBatchReport, ReactiveConfig,
17        ReactiveEngine, ReactiveHandler, ReactiveInterest, ReactiveRuntime, SubscriberBackfill,
18    },
19};
20
21use crate::{
22    ChainlinkFeedProvider, FeedConfig, FeedId, FeedRegistration, OracleAdapterFeedSkip,
23    OracleAdapterPlugin, OracleCodeRegistry, OracleCodeWarmupPolicy, OracleCodeWarmupReport,
24    OracleDiscoveryContext, OracleError, OracleFeedReadinessReport, OracleFeedStatus,
25    OracleHookEvent, OraclePrice, OraclePriceCorrected, OraclePriceUpdate, OracleReactiveHandler,
26    OracleReadOverlay, OracleReconciler, OracleRegistry, OracleSignal, OracleStorageAdapter,
27    OracleStorageSync, OracleTracker, RoundData, StalenessPolicy,
28};
29
30type OracleEventCallback = Arc<dyn Fn(&OracleHookEvent) + Send + Sync>;
31
32alloy_sol_types::sol! {
33    function latestRoundData() external view returns (
34        uint80 roundId,
35        int256 answer,
36        uint256 startedAt,
37        uint256 updatedAt,
38        uint80 answeredInRound
39    );
40}
41
42/// Builder-friendly Chainlink feed registration.
43#[derive(Clone, Debug)]
44pub struct ChainlinkFeed {
45    config: FeedConfig,
46}
47
48impl ChainlinkFeed {
49    /// Create a feed registration for a Chainlink-compatible proxy.
50    pub fn new(proxy: Address) -> Self {
51        Self {
52            config: FeedConfig {
53                proxy,
54                ..Default::default()
55            },
56        }
57    }
58
59    /// Set a stable feed id.
60    pub fn id(mut self, id: impl Into<String>) -> Self {
61        self.config.id = Some(FeedId::new(id));
62        self
63    }
64
65    /// Set a stable feed id.
66    pub fn feed_id(mut self, id: FeedId) -> Self {
67        self.config.id = Some(id);
68        self
69    }
70
71    /// Set a human-readable label.
72    pub fn label(mut self, label: impl Into<String>) -> Self {
73        self.config.label = Some(label.into());
74        self
75    }
76
77    /// Set the base symbol.
78    pub fn base(mut self, base: impl Into<String>) -> Self {
79        self.config.base = Some(base.into());
80        self
81    }
82
83    /// Set the quote symbol.
84    pub fn quote(mut self, quote: impl Into<String>) -> Self {
85        self.config.quote = Some(quote.into());
86        self
87    }
88
89    /// Set a max-age staleness policy.
90    pub fn max_age_secs(mut self, max_age_secs: u64) -> Self {
91        self.config.staleness = StalenessPolicy::max_age(max_age_secs);
92        self
93    }
94
95    /// Set the full staleness policy.
96    pub fn staleness(mut self, staleness: StalenessPolicy) -> Self {
97        self.config.staleness = staleness;
98        self
99    }
100
101    /// Allow zero or negative answers for signed generic feeds.
102    pub fn allow_zero_or_negative_answer(mut self, allow: bool) -> Self {
103        self.config.staleness = self.config.staleness.allow_zero_or_negative_answer(allow);
104        self
105    }
106
107    /// Convert into the lower-level registration config.
108    pub fn into_config(self) -> FeedConfig {
109        self.config
110    }
111}
112
113impl From<ChainlinkFeed> for FeedConfig {
114    fn from(feed: ChainlinkFeed) -> Self {
115        feed.into_config()
116    }
117}
118
119/// Builder for a high-level oracle runtime facade.
120pub struct OracleRuntimeBuilder<P> {
121    provider: P,
122    feeds: Vec<ChainlinkFeed>,
123    now_timestamp: Option<u64>,
124    storage_sync: OracleStorageSync,
125    callbacks: Vec<OracleEventCallback>,
126}
127
128/// Cache-native runtime builder with first-class oracle adapter plugins.
129pub struct OracleCacheRuntimeBuilder {
130    adapters: Vec<Arc<dyn OracleAdapterPlugin>>,
131    now_timestamp: Option<u64>,
132    storage_sync: OracleStorageSync,
133    storage_warmup_enabled: bool,
134    storage_warmup_mode: OracleStorageWarmupMode,
135    code_registry: OracleCodeRegistry,
136    code_warmup_policy: OracleCodeWarmupPolicy,
137    callbacks: Vec<OracleEventCallback>,
138}
139
140/// Best-effort cache-native oracle runtime build report.
141#[non_exhaustive]
142#[derive(Debug)]
143pub struct OracleCacheRuntimeBuildReport {
144    /// Runtime seeded with every successfully discovered feed.
145    pub runtime: OracleRuntime<()>,
146    /// Feeds that were skipped because an adapter could not register them.
147    pub skipped: Vec<OracleAdapterFeedSkip>,
148    /// Feed/runtime readiness after discovery, warmup, and skipped-source classification.
149    pub feed_statuses: Vec<OracleFeedReadinessReport>,
150    /// Storage slots warmed for direct oracle event writes.
151    pub storage_warmup: OracleStorageWarmupReport,
152    /// Bytecode seed/verify/etch results.
153    pub code_warmup: OracleCodeWarmupReport,
154}
155
156/// Result of prewarming oracle storage slots through `evm-fork-cache`.
157#[derive(Clone, Debug, Default, PartialEq, Eq)]
158pub struct OracleStorageWarmupReport {
159    /// Warmup mode used for this run.
160    pub mode: OracleStorageWarmupMode,
161    /// Per-feed readiness after storage warmup.
162    pub feed_statuses: Vec<OracleFeedReadinessReport>,
163    /// Slots requested for warmup.
164    pub requested_slots: usize,
165    /// Slots successfully loaded into the cache.
166    pub loaded_slots: usize,
167    /// Slots that failed to load.
168    pub failed_slots: Vec<OracleStorageWarmupFailure>,
169    /// Read calls submitted to cold-start discovery.
170    pub discovery_calls: usize,
171    /// Cold-start summary when the warmup used `EvmCache::run_cold_start`.
172    pub cold_start: Option<OracleColdStartWarmupReport>,
173}
174
175impl OracleStorageWarmupReport {
176    /// Return true when no storage warmup work was requested or performed.
177    pub fn is_empty(&self) -> bool {
178        self.requested_slots == 0
179            && self.loaded_slots == 0
180            && self.failed_slots.is_empty()
181            && self.discovery_calls == 0
182            && self.cold_start.is_none()
183    }
184}
185
186/// Storage warmup implementation to use for oracle direct-write slots.
187#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
188pub enum OracleStorageWarmupMode {
189    /// Bulk-fetch declared hot slots through `EvmCache::prewarm_slots`.
190    #[default]
191    PrewarmSlots,
192    /// Verify declared hot slots through `EvmCache::run_cold_start`.
193    ColdStart,
194}
195
196/// Summary of an oracle cold-start storage warmup run.
197#[derive(Clone, Debug, Default, PartialEq, Eq)]
198pub struct OracleColdStartWarmupReport {
199    /// Cold-start rounds executed.
200    pub rounds: usize,
201    /// Total slots requested in verify phases.
202    pub verified_slots: usize,
203    /// Slots whose fetched value changed and was injected.
204    pub changed_slots: usize,
205    /// Total fetch failures across verify/probe phases.
206    pub failed_slots: usize,
207    /// Total storage slots discovered by optional discovery calls.
208    pub discovered_slots: usize,
209    /// Total accounts discovered by optional discovery calls.
210    pub discovered_accounts: usize,
211    /// Total discovery calls executed.
212    pub discover_calls: usize,
213}
214
215/// One failed oracle storage warmup slot.
216#[derive(Clone, Debug, PartialEq, Eq)]
217pub struct OracleStorageWarmupFailure {
218    /// Contract address.
219    pub address: Address,
220    /// Storage slot.
221    pub slot: U256,
222    /// Fetch error message.
223    pub reason: String,
224}
225
226/// Result of registering oracle handlers with a reactive engine.
227#[derive(Clone, Debug, Default, PartialEq, Eq)]
228pub struct OracleReactiveInstallReport {
229    /// Handler ids submitted to the engine.
230    pub handler_ids: Vec<HandlerId>,
231}
232
233impl OracleReactiveInstallReport {
234    /// Number of handlers submitted.
235    pub fn len(&self) -> usize {
236        self.handler_ids.len()
237    }
238
239    /// Return true when no handlers were submitted.
240    pub fn is_empty(&self) -> bool {
241        self.handler_ids.is_empty()
242    }
243}
244
245/// Result of unregistering oracle handlers from a reactive engine.
246#[derive(Clone, Debug, Default, PartialEq, Eq)]
247pub struct OracleReactiveUninstallReport {
248    /// Handler ids requested for removal.
249    pub handler_ids: Vec<HandlerId>,
250    /// Handler ids that were present and removed.
251    pub removed_handler_ids: Vec<HandlerId>,
252}
253
254/// Result of refreshing oracle handlers after runtime mutation.
255#[derive(Clone, Debug, Default, PartialEq, Eq)]
256pub struct OracleReactiveRefreshReport {
257    /// Handler ids requested for removal from the previous install handle.
258    pub previous_handler_ids: Vec<HandlerId>,
259    /// Previous handler ids that were present and removed.
260    pub removed_handler_ids: Vec<HandlerId>,
261    /// Current handler ids installed after the refresh.
262    pub installed_handler_ids: Vec<HandlerId>,
263}
264
265/// Result of mutating a runtime's oracle registration set.
266#[derive(Clone, Debug, Default)]
267pub struct OracleRuntimeMutationReport {
268    /// Feed ids registered by this mutation.
269    pub registered_feed_ids: Vec<FeedId>,
270    /// Feed ids removed by this mutation.
271    pub removed_feed_ids: Vec<FeedId>,
272    /// Feeds skipped by adapter discovery, when applicable.
273    pub skipped: Vec<OracleAdapterFeedSkip>,
274    /// Feed/runtime readiness after the mutation.
275    pub feed_statuses: Vec<OracleFeedReadinessReport>,
276    /// Current handler ids after this mutation.
277    pub current_handler_ids: Vec<HandlerId>,
278}
279
280#[derive(Clone)]
281struct OracleAdapterRuntimeState {
282    adapter: Arc<dyn OracleAdapterPlugin>,
283    registrations: Vec<FeedRegistration>,
284}
285
286impl fmt::Debug for OracleAdapterRuntimeState {
287    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
288        f.debug_struct("OracleAdapterRuntimeState")
289            .field("adapter_id", &self.adapter.adapter_id())
290            .field("registrations_len", &self.registrations.len())
291            .finish()
292    }
293}
294
295impl<P> OracleRuntimeBuilder<P> {
296    /// Create a runtime builder around a provider.
297    pub fn new(provider: P) -> Self {
298        Self {
299            provider,
300            feeds: Vec::new(),
301            now_timestamp: None,
302            storage_sync: OracleStorageSync::chainlink_defaults(),
303            callbacks: Vec::new(),
304        }
305    }
306
307    /// Add one feed registration.
308    pub fn feed(mut self, feed: ChainlinkFeed) -> Self {
309        self.feeds.push(feed);
310        self
311    }
312
313    /// Set a fixed timestamp for deterministic registration status classification.
314    pub fn now_timestamp(mut self, now_timestamp: u64) -> Self {
315        self.now_timestamp = Some(now_timestamp);
316        self
317    }
318
319    /// Add one storage adapter.
320    pub fn storage_adapter<A>(mut self, adapter: A) -> Self
321    where
322        A: OracleStorageAdapter + 'static,
323    {
324        self.storage_sync.push_adapter(Arc::new(adapter));
325        self
326    }
327
328    /// Replace the storage-sync registry.
329    pub fn storage_sync(mut self, storage_sync: OracleStorageSync) -> Self {
330        self.storage_sync = storage_sync;
331        self
332    }
333
334    /// Subscribe to every typed oracle event emitted by this runtime.
335    pub fn on_event<F>(mut self, callback: F) -> Self
336    where
337        F: Fn(&OracleHookEvent) + Send + Sync + 'static,
338    {
339        self.callbacks.push(Arc::new(callback));
340        self
341    }
342
343    /// Subscribe to immediate event-derived price updates.
344    pub fn on_price_update<F>(self, callback: F) -> Self
345    where
346        F: Fn(&OraclePriceUpdate) + Send + Sync + 'static,
347    {
348        self.on_event(move |event| {
349            if let OracleHookEvent::PriceUpdate(update) = event {
350                callback(update);
351            }
352        })
353    }
354
355    /// Subscribe to proxy corrections.
356    pub fn on_price_corrected<F>(self, callback: F) -> Self
357    where
358        F: Fn(&OraclePriceCorrected) + Send + Sync + 'static,
359    {
360        self.on_event(move |event| {
361            if let OracleHookEvent::PriceCorrected(corrected) = event {
362                callback(corrected);
363            }
364        })
365    }
366}
367
368impl<P: ChainlinkFeedProvider> OracleRuntimeBuilder<P> {
369    /// Register feeds and construct the runtime.
370    ///
371    /// # Errors
372    ///
373    /// Returns [`OracleError::Provider`] when a proxy read fails,
374    /// [`OracleError::DuplicateFeedId`]/[`OracleError::DuplicateProxy`] when
375    /// two feeds collide, and [`OracleError::Config`] when the system clock is
376    /// before the UNIX epoch and no `now_timestamp` was set.
377    pub async fn build(self) -> Result<OracleRuntime<P>, OracleError> {
378        let now_timestamp = match self.now_timestamp {
379            Some(now_timestamp) => now_timestamp,
380            None => SystemTime::now()
381                .duration_since(UNIX_EPOCH)
382                .map_err(crate::error::clock_error)?
383                .as_secs(),
384        };
385        let mut registry = OracleRegistry::new_at_timestamp(now_timestamp);
386        for feed in self.feeds {
387            registry
388                .register_chainlink_feed(&self.provider, feed.into_config())
389                .await?;
390        }
391
392        Ok(OracleRuntime {
393            provider: self.provider,
394            tracker: OracleTracker::new(registry),
395            storage_sync: self.storage_sync,
396            callbacks: self.callbacks,
397            adapter_states: Vec::new(),
398        })
399    }
400}
401
402impl Default for OracleCacheRuntimeBuilder {
403    fn default() -> Self {
404        Self {
405            adapters: Vec::new(),
406            now_timestamp: None,
407            storage_sync: OracleStorageSync::chainlink_defaults(),
408            storage_warmup_enabled: true,
409            storage_warmup_mode: OracleStorageWarmupMode::default(),
410            code_registry: OracleCodeRegistry::default(),
411            code_warmup_policy: OracleCodeWarmupPolicy::default(),
412            callbacks: Vec::new(),
413        }
414    }
415}
416
417impl OracleCacheRuntimeBuilder {
418    /// Install one oracle adapter plugin.
419    pub fn install_adapter<A>(mut self, adapter: A) -> Self
420    where
421        A: OracleAdapterPlugin + 'static,
422    {
423        self.adapters.push(Arc::new(adapter));
424        self
425    }
426
427    /// Set a fixed timestamp for deterministic registration status classification.
428    pub fn now_timestamp(mut self, now_timestamp: u64) -> Self {
429        self.now_timestamp = Some(now_timestamp);
430        self
431    }
432
433    /// Add one storage adapter.
434    pub fn storage_adapter<A>(mut self, adapter: A) -> Self
435    where
436        A: OracleStorageAdapter + 'static,
437    {
438        self.storage_sync.push_adapter(Arc::new(adapter));
439        self
440    }
441
442    /// Replace the storage-sync registry.
443    pub fn storage_sync(mut self, storage_sync: OracleStorageSync) -> Self {
444        self.storage_sync = storage_sync;
445        self
446    }
447
448    /// Enable or disable oracle storage warmup during cache-native builds.
449    ///
450    /// Enabled by default. When enabled, the builder asks installed storage
451    /// adapters which slots must be hot for direct event writes and bulk-loads
452    /// them through the cache before returning the runtime.
453    pub fn storage_warmup(mut self, enabled: bool) -> Self {
454        self.storage_warmup_enabled = enabled;
455        self
456    }
457
458    /// Set the oracle storage warmup mode.
459    pub fn storage_warmup_mode(mut self, mode: OracleStorageWarmupMode) -> Self {
460        self.storage_warmup_mode = mode;
461        self
462    }
463
464    /// Use `EvmCache::run_cold_start` for declared oracle storage slots.
465    pub fn storage_cold_start(self) -> Self {
466        self.storage_warmup_mode(OracleStorageWarmupMode::ColdStart)
467    }
468
469    /// Disable oracle storage warmup.
470    pub fn disable_storage_warmup(self) -> Self {
471        self.storage_warmup(false)
472    }
473
474    /// Replace the bytecode seed/etch registry.
475    pub fn code_registry(mut self, registry: OracleCodeRegistry) -> Self {
476        self.code_registry = registry;
477        self
478    }
479
480    /// Set the bytecode warmup failure policy.
481    pub fn code_warmup_policy(mut self, policy: OracleCodeWarmupPolicy) -> Self {
482        self.code_warmup_policy = policy;
483        self
484    }
485
486    /// Add a canonical bytecode seed that must verify against on-chain code hash.
487    pub fn code_seed(mut self, address: Address, code: Bytes) -> Self {
488        self.code_registry = self.code_registry.seed(address, code);
489        self
490    }
491
492    /// Add multiple canonical bytecode seeds that must verify against chain code hash.
493    pub fn code_seeds<I>(mut self, seeds: I) -> Self
494    where
495        I: IntoIterator<Item = (Address, Bytes)>,
496    {
497        self.code_registry = self.code_registry.seed_many(seeds);
498        self
499    }
500
501    /// Add explicit simulation-only bytecode at `address`.
502    pub fn code_etch(mut self, address: Address, code: Bytes) -> Self {
503        self.code_registry = self.code_registry.etch(address, code);
504        self
505    }
506
507    /// Add multiple explicit simulation-only bytecode etches.
508    pub fn code_etches<I>(mut self, etches: I) -> Self
509    where
510        I: IntoIterator<Item = (Address, Bytes)>,
511    {
512        self.code_registry = self.code_registry.etch_many(etches);
513        self
514    }
515
516    /// Subscribe to every typed oracle event emitted by this runtime.
517    pub fn on_event<F>(mut self, callback: F) -> Self
518    where
519        F: Fn(&OracleHookEvent) + Send + Sync + 'static,
520    {
521        self.callbacks.push(Arc::new(callback));
522        self
523    }
524
525    /// Subscribe to immediate event-derived price updates.
526    pub fn on_price_update<F>(self, callback: F) -> Self
527    where
528        F: Fn(&OraclePriceUpdate) + Send + Sync + 'static,
529    {
530        self.on_event(move |event| {
531            if let OracleHookEvent::PriceUpdate(update) = event {
532                callback(update);
533            }
534        })
535    }
536
537    /// Register adapter-discovered feeds and construct the runtime.
538    ///
539    /// # Errors
540    ///
541    /// Returns [`OracleError::FeedSkipped`] when any adapter reported a
542    /// skipped feed (use [`Self::build_report`] to receive skips as data),
543    /// [`OracleError::Policy`] when code warmup fails the configured
544    /// [`OracleCodeWarmupPolicy`], [`OracleError::Provider`] when adapter
545    /// discovery reads fail, [`OracleError::DuplicateFeedId`]/
546    /// [`OracleError::DuplicateProxy`] when discovered feeds collide, and
547    /// [`OracleError::Config`] when the system clock is before the UNIX epoch
548    /// and no `now_timestamp` was set.
549    pub async fn build(self, cache: &mut EvmCache) -> Result<OracleRuntime<()>, OracleError> {
550        let report = self.build_report(cache).await?;
551        if let Some(skipped) = report.skipped.first() {
552            return Err(cache_runtime_skip_error(skipped));
553        }
554        Ok(report.runtime)
555    }
556
557    /// Register every compatible adapter feed and return skipped feeds instead of failing fast.
558    pub async fn build_report(
559        self,
560        cache: &mut EvmCache,
561    ) -> Result<OracleCacheRuntimeBuildReport, OracleError> {
562        let now_timestamp = match self.now_timestamp {
563            Some(now_timestamp) => now_timestamp,
564            None => SystemTime::now()
565                .duration_since(UNIX_EPOCH)
566                .map_err(crate::error::clock_error)?
567                .as_secs(),
568        };
569
570        let code_warmup = self
571            .code_registry
572            .apply_to_cache_with_policy(cache, self.code_warmup_policy)?;
573        let mut discovered_by_adapter = Vec::new();
574        let mut seeded = Vec::new();
575        let mut warmup_registrations = Vec::new();
576        let mut skipped = Vec::new();
577        for adapter in &self.adapters {
578            let report = adapter
579                .discover(OracleDiscoveryContext {
580                    cache,
581                    now_timestamp,
582                })
583                .await?;
584            skipped.extend(report.skipped);
585            let registrations = report
586                .feeds
587                .iter()
588                .map(|feed| feed.registration.clone())
589                .collect::<Vec<_>>();
590            seeded.extend(report.feeds.into_iter().map(|feed| {
591                warmup_registrations.push(feed.registration.clone());
592                (feed.registration, feed.round)
593            }));
594            discovered_by_adapter.push(OracleAdapterRuntimeState {
595                adapter: Arc::clone(adapter),
596                registrations,
597            });
598        }
599
600        let storage_warmup = if self.storage_warmup_enabled {
601            prewarm_oracle_storage(
602                cache,
603                &self.storage_sync,
604                &warmup_registrations.iter().collect::<Vec<_>>(),
605                self.storage_warmup_mode,
606            )
607        } else {
608            OracleStorageWarmupReport {
609                feed_statuses: feed_readiness_from_registrations(&warmup_registrations),
610                ..OracleStorageWarmupReport::default()
611            }
612        };
613        let mut feed_statuses = storage_warmup.feed_statuses.clone();
614        feed_statuses.extend(skipped.iter().map(skipped_feed_readiness));
615        let tracker = OracleTracker::from_registrations_at_timestamp(seeded, now_timestamp)?;
616        Ok(OracleCacheRuntimeBuildReport {
617            runtime: OracleRuntime {
618                provider: (),
619                tracker,
620                storage_sync: self.storage_sync,
621                callbacks: self.callbacks,
622                adapter_states: discovered_by_adapter,
623            },
624            skipped,
625            feed_statuses,
626            storage_warmup,
627            code_warmup,
628        })
629    }
630}
631
632/// High-level facade for registration, typed events, reads, and reconciliation.
633pub struct OracleRuntime<P> {
634    provider: P,
635    tracker: OracleTracker,
636    storage_sync: OracleStorageSync,
637    callbacks: Vec<OracleEventCallback>,
638    adapter_states: Vec<OracleAdapterRuntimeState>,
639}
640
641impl<P: fmt::Debug> fmt::Debug for OracleRuntime<P> {
642    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
643        f.debug_struct("OracleRuntime")
644            .field("provider", &self.provider)
645            .field("tracker", &self.tracker)
646            .field("storage_sync", &self.storage_sync)
647            .field("callbacks_len", &self.callbacks.len())
648            .field("adapter_states", &self.adapter_states)
649            .finish()
650    }
651}
652
653impl<P> OracleRuntime<P> {
654    /// Start building a runtime around a provider.
655    pub fn builder(provider: P) -> OracleRuntimeBuilder<P> {
656        OracleRuntimeBuilder::new(provider)
657    }
658
659    /// Borrow the provider used for authoritative proxy reads and reconciliation.
660    pub fn provider(&self) -> &P {
661        &self.provider
662    }
663
664    /// Borrow the typed state store for snapshot, price, and registration reads.
665    pub fn tracker(&self) -> &OracleTracker {
666        &self.tracker
667    }
668
669    /// Mutably borrow the tracker.
670    ///
671    /// **Warning:** mutating the tracker directly (registering or removing
672    /// feeds through [`OracleTracker`] methods) desynchronizes this runtime's
673    /// adapter bookkeeping: `adapter_states` still references removed feeds
674    /// (or misses added ones), so [`Self::reactive_handlers`] and the
675    /// `refresh_handlers*` methods rebuild handler routing from a stale feed
676    /// set. Use [`Self::register_seeded_feed`], [`Self::register_adapter`],
677    /// [`Self::unregister_feed_by_id`], or [`Self::unregister_feed_by_proxy`]
678    /// instead — they keep typed state and handler routing in sync. Reserve
679    /// this accessor for operations that do not change the registration set
680    /// (for example [`OracleTracker::reconcile`] or
681    /// [`OracleTracker::apply_batch_report`]).
682    pub fn tracker_mut(&mut self) -> &mut OracleTracker {
683        &mut self.tracker
684    }
685
686    /// Return feed/runtime readiness for all registered feeds.
687    pub fn feed_readiness(&self) -> Vec<OracleFeedReadinessReport> {
688        self.tracker.feed_readiness()
689    }
690
691    pub(crate) fn storage_sync(&self) -> &OracleStorageSync {
692        &self.storage_sync
693    }
694
695    /// Replace the direct-storage registry used by subsequently constructed
696    /// reactive handlers and storage warmup calls.
697    ///
698    /// This is primarily useful after cache-native discovery returns an
699    /// [`OracleTracker`] that needs a deployment-specific storage adapter:
700    /// construct the facade with [`OracleRuntime::from_tracker`], then retain
701    /// the custom registry here before calling [`Self::reactive_runtime`],
702    /// [`Self::reactive_handler`], or [`Self::prewarm_storage`].
703    ///
704    /// If this runtime's handlers are already installed in a live reactive
705    /// engine, replace the registry only as part of a handler refresh so the
706    /// installed handlers cannot retain the old storage policy.
707    pub fn with_storage_sync(mut self, storage_sync: OracleStorageSync) -> Self {
708        self.storage_sync = storage_sync;
709        self
710    }
711
712    /// Build a fresh built-in Chainlink-compatible reactive handler from current registrations.
713    ///
714    /// Adapter plugin handlers are not included. Use [`Self::reactive_handlers`],
715    /// [`Self::reactive_runtime`], or [`Self::register_subscriber`] when the runtime
716    /// was built with installed adapter plugins.
717    pub fn reactive_handler(&self) -> OracleReactiveHandler {
718        OracleReactiveHandler::with_storage_sync(
719            self.tracker.registrations().collect(),
720            self.storage_sync.clone(),
721        )
722    }
723
724    /// Build fresh reactive handlers for every source installed in this runtime.
725    ///
726    /// The first handler is the built-in Chainlink-compatible handler. Any adapter
727    /// plugin handlers discovered during cache-native startup follow it.
728    ///
729    /// Handler ids must be unique across installed adapter plugins: when two
730    /// adapters return handlers with the same [`HandlerId`], only the first
731    /// handler is kept and the second adapter's event routing is silently
732    /// dropped (a `tracing` warning is emitted). An adapter handler that
733    /// reuses the built-in Chainlink handler id is fine — the built-in
734    /// handler is rebuilt from *all* current registrations, so it already
735    /// routes that adapter's feeds.
736    pub fn reactive_handlers(&self) -> Vec<Arc<dyn ReactiveHandler<Ethereum>>> {
737        let built_in: Arc<dyn ReactiveHandler<Ethereum>> = Arc::new(self.reactive_handler());
738        let built_in_id = built_in.id();
739        let mut adapter_ids = BTreeSet::new();
740        let mut handlers = vec![built_in];
741        for state in self
742            .adapter_states
743            .iter()
744            .filter(|state| !state.registrations.is_empty())
745        {
746            let handler = state
747                .adapter
748                .reactive_handler(state.registrations.clone(), self.storage_sync.clone());
749            let id = handler.id();
750            if id == built_in_id {
751                // By design: the adapter delegates to the built-in Chainlink
752                // handler, which is already built from all registrations.
753                continue;
754            }
755            if !adapter_ids.insert(id.clone()) {
756                tracing::warn!(
757                    handler_id = %id,
758                    adapter_id = %state.adapter.adapter_id(),
759                    "duplicate reactive HandlerId from adapter plugins; \
760                     dropping this adapter's handler routing"
761                );
762                continue;
763            }
764            handlers.push(handler);
765        }
766        handlers
767    }
768
769    /// Return current reactive handler ids for this runtime.
770    pub fn reactive_handler_ids(&self) -> Vec<HandlerId> {
771        self.reactive_handlers()
772            .into_iter()
773            .map(|handler| handler.id())
774            .collect()
775    }
776
777    /// Build a reactive runtime with this oracle runtime's handlers already installed.
778    pub fn reactive_runtime(&self) -> Result<ReactiveRuntime<Ethereum>, OracleError> {
779        let mut runtime = ReactiveRuntime::<Ethereum>::new(ReactiveConfig::default());
780        for handler in self.reactive_handlers() {
781            runtime.register_handler(handler).map_err(runtime_error)?;
782        }
783        Ok(runtime)
784    }
785
786    /// Register this runtime's handlers with an engine, using continuity-safe backfill.
787    ///
788    /// # Errors
789    ///
790    /// Returns [`OracleError::Reactive`] when the engine rejects a handler
791    /// registration (for example a handler id that is already installed).
792    pub fn install_handlers<S>(
793        &self,
794        engine: &mut ReactiveEngine<S, Ethereum>,
795    ) -> Result<OracleReactiveInstallReport, OracleError>
796    where
797        S: InterestOwnerSubscriber<Ethereum>,
798    {
799        let mut report = OracleReactiveInstallReport::default();
800        for handler in self.reactive_handlers() {
801            let id = handler.id();
802            engine.register_handler(handler).map_err(runtime_error)?;
803            report.handler_ids.push(id);
804        }
805        Ok(report)
806    }
807
808    /// Refresh an existing engine install after registering or unregistering feeds.
809    ///
810    /// `installed` must be the report returned by a previous install/refresh.
811    /// The old handler ids in that report are removed first, then the runtime's
812    /// current handlers are installed with continuity-safe backfill.
813    pub fn refresh_handlers<S>(
814        &self,
815        engine: &mut ReactiveEngine<S, Ethereum>,
816        installed: &mut OracleReactiveInstallReport,
817    ) -> Result<OracleReactiveRefreshReport, OracleError>
818    where
819        S: InterestOwnerSubscriber<Ethereum>,
820    {
821        self.refresh_handlers_inner(engine, installed, OracleHandlerRefreshBackfill::Continuity)
822    }
823
824    /// Refresh an existing engine install after mutation using explicit backfill.
825    pub fn refresh_handlers_with_backfill<S>(
826        &self,
827        engine: &mut ReactiveEngine<S, Ethereum>,
828        installed: &mut OracleReactiveInstallReport,
829        backfill: SubscriberBackfill,
830    ) -> Result<OracleReactiveRefreshReport, OracleError>
831    where
832        S: InterestOwnerSubscriber<Ethereum>,
833    {
834        self.refresh_handlers_inner(
835            engine,
836            installed,
837            OracleHandlerRefreshBackfill::Explicit(backfill),
838        )
839    }
840
841    /// Refresh an existing engine install after mutation without backfill.
842    pub fn refresh_handlers_live_only<S>(
843        &self,
844        engine: &mut ReactiveEngine<S, Ethereum>,
845        installed: &mut OracleReactiveInstallReport,
846    ) -> Result<OracleReactiveRefreshReport, OracleError>
847    where
848        S: InterestOwnerSubscriber<Ethereum>,
849    {
850        self.refresh_handlers_inner(engine, installed, OracleHandlerRefreshBackfill::LiveOnly)
851    }
852
853    /// Unregister exactly the handlers recorded by a previous install/refresh report.
854    pub fn uninstall_installed_handlers<S>(
855        &self,
856        engine: &mut ReactiveEngine<S, Ethereum>,
857        installed: &mut OracleReactiveInstallReport,
858    ) -> OracleReactiveUninstallReport
859    where
860        S: InterestOwnerSubscriber<Ethereum>,
861    {
862        let handler_ids = std::mem::take(&mut installed.handler_ids);
863        let removed_handler_ids = unregister_handler_ids(engine, &handler_ids);
864        OracleReactiveUninstallReport {
865            handler_ids,
866            removed_handler_ids,
867        }
868    }
869
870    /// Register this runtime's handlers with explicit owner-scoped backfill.
871    ///
872    /// # Errors
873    ///
874    /// Returns [`OracleError::Reactive`] when the engine rejects a handler
875    /// registration (for example a handler id that is already installed).
876    pub fn install_handlers_with_backfill<S>(
877        &self,
878        engine: &mut ReactiveEngine<S, Ethereum>,
879        backfill: SubscriberBackfill,
880    ) -> Result<OracleReactiveInstallReport, OracleError>
881    where
882        S: InterestOwnerSubscriber<Ethereum>,
883    {
884        let mut report = OracleReactiveInstallReport::default();
885        for handler in self.reactive_handlers() {
886            let id = handler.id();
887            engine
888                .register_handler_with_backfill(handler, backfill)
889                .map_err(runtime_error)?;
890            report.handler_ids.push(id);
891        }
892        Ok(report)
893    }
894
895    /// Register this runtime's handlers without any backfill.
896    ///
897    /// # Errors
898    ///
899    /// Returns [`OracleError::Reactive`] when the engine rejects a handler
900    /// registration (for example a handler id that is already installed).
901    pub fn install_handlers_live_only<S>(
902        &self,
903        engine: &mut ReactiveEngine<S, Ethereum>,
904    ) -> Result<OracleReactiveInstallReport, OracleError>
905    where
906        S: InterestOwnerSubscriber<Ethereum>,
907    {
908        let mut report = OracleReactiveInstallReport::default();
909        for handler in self.reactive_handlers() {
910            let id = handler.id();
911            engine
912                .register_handler_live_only(handler)
913                .map_err(runtime_error)?;
914            report.handler_ids.push(id);
915        }
916        Ok(report)
917    }
918
919    /// Unregister this runtime's handlers from an engine.
920    ///
921    /// **Warning:** this recomputes handler ids from the *current*
922    /// registration set. If the runtime's registrations changed since the
923    /// handlers were installed (feeds registered or unregistered, adapters
924    /// added), the recomputed set can miss handlers that are still installed
925    /// in the engine, leaving them registered. Prefer
926    /// [`Self::uninstall_installed_handlers`] with the report returned by the
927    /// original install/refresh call — it removes exactly the handler ids
928    /// that were installed.
929    pub fn uninstall_handlers<S>(
930        &self,
931        engine: &mut ReactiveEngine<S, Ethereum>,
932    ) -> OracleReactiveUninstallReport
933    where
934        S: InterestOwnerSubscriber<Ethereum>,
935    {
936        let mut report = OracleReactiveUninstallReport::default();
937        for handler in self.reactive_handlers() {
938            let id = handler.id();
939            report.handler_ids.push(id.clone());
940            if engine.unregister_handler(&id).is_some() {
941                report.removed_handler_ids.push(id);
942            }
943        }
944        report
945    }
946
947    /// Register this runtime's oracle interests with an event subscriber.
948    pub fn register_subscriber<S>(&self, subscriber: &mut S) -> Result<(), OracleError>
949    where
950        S: EventSubscriber<Ethereum>,
951    {
952        subscriber
953            .register_interests(&self.reactive_interests())
954            .map_err(runtime_error)
955    }
956
957    /// Read one subscriber batch, apply it to the cache, and return the typed
958    /// batch digest, or `None` when the subscriber has no batch ready.
959    pub async fn next_subscriber_events<S>(
960        &mut self,
961        cache: &mut EvmCache,
962        runtime: &mut ReactiveRuntime<Ethereum>,
963        subscriber: &mut S,
964    ) -> Result<Option<crate::OracleBatchReport>, OracleError>
965    where
966        S: EventSubscriber<Ethereum>,
967    {
968        let Some(batch) = subscriber.next_batch().await.map_err(runtime_error)? else {
969            return Ok(None);
970        };
971        let report = runtime.ingest_batch(cache, batch).map_err(runtime_error)?;
972        self.apply_batch_report(&report).map(Some)
973    }
974
975    /// Build a read overlay over the current tracker state.
976    pub fn read_overlay(&self) -> OracleReadOverlay<'_> {
977        OracleReadOverlay::new(&self.tracker)
978    }
979
980    /// Register an already-discovered feed and seed its latest round.
981    ///
982    /// This updates typed state immediately. If a reactive engine has already
983    /// installed this runtime's handlers, call one of the `refresh_handlers*`
984    /// methods with the previous install report so the subscriber receives the
985    /// new event interests.
986    ///
987    /// # Errors
988    ///
989    /// Returns [`OracleError::DuplicateFeedId`] or
990    /// [`OracleError::DuplicateProxy`] when the registration collides with an
991    /// existing feed; typed state is unchanged in that case.
992    pub fn register_seeded_feed(
993        &mut self,
994        registration: FeedRegistration,
995        round: RoundData,
996    ) -> Result<OracleRuntimeMutationReport, OracleError> {
997        let id = registration.id.clone();
998        self.tracker
999            .insert_seeded_registration(registration, round)?;
1000        Ok(self.mutation_report([id], []))
1001    }
1002
1003    /// Register an already-discovered adapter-owned feed and seed its latest round.
1004    ///
1005    /// The adapter must already be installed in this runtime. This is the
1006    /// low-level path for adapter code that discovers one additional feed and
1007    /// wants that feed included in the adapter handler rebuilt on refresh.
1008    ///
1009    /// # Errors
1010    ///
1011    /// Returns [`OracleError::Config`] when `adapter_id` is not installed in
1012    /// this runtime, and [`OracleError::DuplicateFeedId`]/
1013    /// [`OracleError::DuplicateProxy`] when the registration collides with an
1014    /// existing feed.
1015    pub fn register_adapter_seeded_feed(
1016        &mut self,
1017        adapter_id: impl AsRef<str>,
1018        registration: FeedRegistration,
1019        round: RoundData,
1020    ) -> Result<OracleRuntimeMutationReport, OracleError> {
1021        let id = registration.id.clone();
1022        let state_index = self
1023            .adapter_state_index(adapter_id.as_ref())
1024            .ok_or_else(|| adapter_not_installed(adapter_id.as_ref()))?;
1025        self.tracker
1026            .insert_seeded_registration(registration.clone(), round)?;
1027        self.adapter_states[state_index]
1028            .registrations
1029            .push(registration);
1030        Ok(self.mutation_report([id], []))
1031    }
1032
1033    /// Unregister a feed by id.
1034    ///
1035    /// This removes typed state, pending reconciliation, and any adapter routing
1036    /// references for the feed. Refresh installed handlers afterward to apply
1037    /// the routing change to a live engine.
1038    ///
1039    /// # Errors
1040    ///
1041    /// Returns [`OracleError::FeedNotFound`] when no feed is registered under
1042    /// `id`.
1043    pub fn unregister_feed_by_id(
1044        &mut self,
1045        id: FeedId,
1046    ) -> Result<OracleRuntimeMutationReport, OracleError> {
1047        let removed = self
1048            .tracker
1049            .remove_by_id(id)
1050            .ok_or(OracleError::FeedNotFound)?;
1051        let id = removed.id.clone();
1052        self.remove_adapter_registration(id.clone());
1053        Ok(self.mutation_report([], [id]))
1054    }
1055
1056    /// Unregister a feed by proxy.
1057    ///
1058    /// # Errors
1059    ///
1060    /// Returns [`OracleError::FeedNotFound`] when no feed is registered under
1061    /// `proxy`.
1062    pub fn unregister_feed_by_proxy(
1063        &mut self,
1064        proxy: Address,
1065    ) -> Result<OracleRuntimeMutationReport, OracleError> {
1066        let removed = self
1067            .tracker
1068            .remove_by_proxy(proxy)
1069            .ok_or(OracleError::FeedNotFound)?;
1070        let id = removed.id.clone();
1071        self.remove_adapter_registration(id.clone());
1072        Ok(self.mutation_report([], [id]))
1073    }
1074
1075    /// Return the latest typed price by feed id string.
1076    ///
1077    /// # Errors
1078    ///
1079    /// Returns [`OracleError::FeedNotFound`] when no feed is registered under
1080    /// `id` or the feed has no current snapshot.
1081    pub fn price(&self, id: impl AsRef<str>) -> Result<OraclePrice, OracleError> {
1082        self.tracker.price(id)
1083    }
1084
1085    /// Return the latest typed price by proxy address.
1086    ///
1087    /// # Errors
1088    ///
1089    /// Returns [`OracleError::FeedNotFound`] when no feed is registered under
1090    /// `proxy` or the feed has no current snapshot.
1091    pub fn price_by_proxy(&self, proxy: Address) -> Result<OraclePrice, OracleError> {
1092        self.tracker.price_by_proxy(proxy)
1093    }
1094
1095    /// Return the latest round by feed id string.
1096    ///
1097    /// # Errors
1098    ///
1099    /// Returns [`OracleError::FeedNotFound`] when no feed is registered under
1100    /// `id` or the feed has no current snapshot.
1101    pub fn latest_round(&self, id: impl AsRef<str>) -> Result<RoundData, OracleError> {
1102        self.tracker.latest_round(id)
1103    }
1104
1105    /// Apply a committed reactive batch and return the typed batch digest.
1106    ///
1107    /// The returned [`crate::OracleBatchReport`] carries the raw per-event
1108    /// hooks (`events`, in emission order) plus the digested per-feed changes,
1109    /// continuity incidents, and the `requires_full_refresh` routing flag —
1110    /// the same consumer shape as `evm-amm-state`'s `AmmSyncBatchReport`.
1111    /// Registered callbacks fire for every event before this returns.
1112    ///
1113    /// Legacy `oracle.answer_updated` signals (emitted by custom handlers that
1114    /// predate the rich [`OraclePriceUpdate`] payload) update typed tracker
1115    /// state but are **not** decoded into [`OracleHookEvent`]s, so `on_event`/
1116    /// `on_price_update` callbacks do not fire for them. Emit the rich
1117    /// `oracle.price_update` signal from custom handlers to get typed
1118    /// callbacks.
1119    pub fn apply_batch_report<N: Network>(
1120        &mut self,
1121        report: &ReactiveBatchReport<N>,
1122    ) -> Result<crate::OracleBatchReport, OracleError> {
1123        let mut events = Vec::new();
1124        for signal in report
1125            .applied
1126            .iter()
1127            .flat_map(|applied| applied.hook_signals.iter())
1128        {
1129            if let Some(signal) = OracleSignal::from_hook(signal)? {
1130                events.push(signal.to_event());
1131            }
1132        }
1133
1134        self.tracker.apply_batch_report(report)?;
1135        self.emit(&events);
1136        Ok(crate::OracleBatchReport::from_events(events))
1137    }
1138
1139    /// Reconcile all currently pending proxy-kind event updates.
1140    ///
1141    /// Derived-source requests
1142    /// ([`crate::OracleReconciliationKind::DerivedProtocolRead`]) are left
1143    /// queued; satisfy those with [`OracleRuntime::reconcile_derived`].
1144    pub async fn reconcile_pending(&mut self) -> Result<Vec<OracleHookEvent>, OracleError>
1145    where
1146        P: ChainlinkFeedProvider,
1147    {
1148        let mut reconciler = OracleReconciler::default();
1149        for request in self
1150            .tracker
1151            .pending_reconciliations()
1152            .iter()
1153            .filter(|request| request.kind == crate::OracleReconciliationKind::Proxy)
1154            .cloned()
1155        {
1156            reconciler.enqueue(request);
1157        }
1158
1159        let mut events = Vec::new();
1160        while let Some(result) = reconciler
1161            .reconcile_next(&mut self.tracker, &self.provider)
1162            .await?
1163        {
1164            events.extend(result.hooks);
1165        }
1166        self.emit(&events);
1167        Ok(events)
1168    }
1169
1170    /// Reconcile all currently pending derived-source updates through their
1171    /// protocols' own view calls (Morpho `price()`, Euler `getQuote`), read
1172    /// through `cache`.
1173    ///
1174    /// Successful reads promote `EventPending` snapshots to `Confirmed` or
1175    /// `Corrected` and fire the registered hook callbacks; failed reads leave
1176    /// their requests queued and are surfaced in
1177    /// [`crate::DerivedReconcileReport::failed`].
1178    pub fn reconcile_derived(
1179        &mut self,
1180        cache: &mut evm_fork_cache::cache::EvmCache,
1181    ) -> crate::DerivedReconcileReport {
1182        let report = self.tracker.reconcile_derived_pending_with(cache);
1183        let events: Vec<OracleHookEvent> = report
1184            .reconciled
1185            .iter()
1186            .flat_map(|result| result.hooks.iter().cloned())
1187            .collect();
1188        self.emit(&events);
1189        report
1190    }
1191
1192    fn emit(&self, events: &[OracleHookEvent]) {
1193        for event in events {
1194            for callback in &self.callbacks {
1195                callback(event);
1196            }
1197        }
1198    }
1199
1200    /// Return the complete log/event interest set for built-in and adapter handlers.
1201    pub fn reactive_interests(&self) -> Vec<ReactiveInterest<Ethereum>> {
1202        let mut interests = Vec::new();
1203        for handler in self.reactive_handlers() {
1204            interests.extend(handler.interests());
1205        }
1206        interests
1207    }
1208
1209    /// Return the oracle storage slots this runtime can prewarm for direct event writes.
1210    pub fn storage_warmup_slots(&self) -> Vec<(Address, U256)> {
1211        self.storage_sync
1212            .warm_slots_for_registrations(self.tracker.registrations_iter())
1213    }
1214
1215    /// Bulk-load direct-write oracle storage slots into the supplied cache.
1216    pub fn prewarm_storage(&self, cache: &mut EvmCache) -> OracleStorageWarmupReport {
1217        let registrations = self.tracker.registrations_iter().collect::<Vec<_>>();
1218        prewarm_oracle_storage(
1219            cache,
1220            &self.storage_sync,
1221            &registrations,
1222            OracleStorageWarmupMode::default(),
1223        )
1224    }
1225
1226    /// Verify direct-write oracle storage slots through `EvmCache::run_cold_start`.
1227    pub fn cold_start_storage(&self, cache: &mut EvmCache) -> OracleStorageWarmupReport {
1228        let registrations = self.tracker.registrations_iter().collect::<Vec<_>>();
1229        prewarm_oracle_storage(
1230            cache,
1231            &self.storage_sync,
1232            &registrations,
1233            OracleStorageWarmupMode::ColdStart,
1234        )
1235    }
1236
1237    fn refresh_handlers_inner<S>(
1238        &self,
1239        engine: &mut ReactiveEngine<S, Ethereum>,
1240        installed: &mut OracleReactiveInstallReport,
1241        backfill: OracleHandlerRefreshBackfill,
1242    ) -> Result<OracleReactiveRefreshReport, OracleError>
1243    where
1244        S: InterestOwnerSubscriber<Ethereum>,
1245    {
1246        let previous_handler_ids = std::mem::take(&mut installed.handler_ids);
1247        let removed_handler_ids = unregister_handler_ids(engine, &previous_handler_ids);
1248        let new_install = match backfill {
1249            OracleHandlerRefreshBackfill::Continuity => self.install_handlers(engine)?,
1250            OracleHandlerRefreshBackfill::Explicit(backfill) => {
1251                self.install_handlers_with_backfill(engine, backfill)?
1252            }
1253            OracleHandlerRefreshBackfill::LiveOnly => self.install_handlers_live_only(engine)?,
1254        };
1255        installed.handler_ids = new_install.handler_ids.clone();
1256        Ok(OracleReactiveRefreshReport {
1257            previous_handler_ids,
1258            removed_handler_ids,
1259            installed_handler_ids: installed.handler_ids.clone(),
1260        })
1261    }
1262
1263    fn mutation_report(
1264        &self,
1265        registered_feed_ids: impl IntoIterator<Item = FeedId>,
1266        removed_feed_ids: impl IntoIterator<Item = FeedId>,
1267    ) -> OracleRuntimeMutationReport {
1268        OracleRuntimeMutationReport {
1269            registered_feed_ids: registered_feed_ids.into_iter().collect(),
1270            removed_feed_ids: removed_feed_ids.into_iter().collect(),
1271            skipped: Vec::new(),
1272            feed_statuses: feed_readiness_from_registrations(self.tracker.registrations_iter()),
1273            current_handler_ids: self.reactive_handler_ids(),
1274        }
1275    }
1276
1277    fn adapter_state_index(&self, adapter_id: &str) -> Option<usize> {
1278        self.adapter_states
1279            .iter()
1280            .position(|state| state.adapter.adapter_id().as_str() == adapter_id)
1281    }
1282
1283    fn push_adapter_state(
1284        &mut self,
1285        adapter: Arc<dyn OracleAdapterPlugin>,
1286        registrations: Vec<FeedRegistration>,
1287    ) {
1288        if registrations.is_empty() {
1289            return;
1290        }
1291
1292        let adapter_id = adapter.adapter_id();
1293        if let Some(existing) = self
1294            .adapter_states
1295            .iter_mut()
1296            .find(|state| state.adapter.adapter_id().as_str() == adapter_id.as_str())
1297        {
1298            existing.registrations.extend(registrations);
1299            return;
1300        }
1301
1302        self.adapter_states.push(OracleAdapterRuntimeState {
1303            adapter,
1304            registrations,
1305        });
1306    }
1307
1308    fn remove_adapter_registration(&mut self, id: FeedId) {
1309        for state in &mut self.adapter_states {
1310            state
1311                .registrations
1312                .retain(|registration| registration.id != id);
1313        }
1314    }
1315}
1316
1317impl OracleRuntime<()> {
1318    /// Start building a cache-native runtime from oracle adapter plugins.
1319    pub fn cache_builder() -> OracleCacheRuntimeBuilder {
1320        OracleCacheRuntimeBuilder::default()
1321    }
1322
1323    /// Build a runtime facade from a cache-native tracker.
1324    ///
1325    /// The returned runtime starts with **Chainlink-default storage sync and
1326    /// no adapter-plugin routing**, regardless of how the tracker's feeds
1327    /// were discovered: any custom [`OracleStorageSync`] configuration is
1328    /// reset to [`OracleStorageSync::chainlink_defaults`] unless replaced with
1329    /// [`OracleRuntime::with_storage_sync`], and non-Chainlink
1330    /// feeds already in the tracker (Pyth, RedStone, custom adapter families)
1331    /// remain visible in typed state but receive **no reactive events** —
1332    /// their adapter handlers are gone until the adapters are re-installed
1333    /// via [`Self::register_adapter`]. Prefer keeping the runtime returned by
1334    /// [`OracleCacheRuntimeBuilder::build`] when adapter plugins are in play.
1335    pub fn from_tracker(tracker: OracleTracker) -> Self {
1336        Self {
1337            provider: (),
1338            tracker,
1339            storage_sync: OracleStorageSync::chainlink_defaults(),
1340            callbacks: Vec::new(),
1341            adapter_states: Vec::new(),
1342        }
1343    }
1344
1345    /// Discover and register feeds from an adapter on an existing cache-native runtime.
1346    ///
1347    /// This mutates typed state and retains the adapter for future handler
1348    /// rebuilds. If handlers are already installed in an engine, call one of the
1349    /// `refresh_handlers*` methods with the previous install report afterward.
1350    ///
1351    /// # Errors
1352    ///
1353    /// Returns [`OracleError::Provider`] when adapter discovery reads fail,
1354    /// and [`OracleError::DuplicateFeedId`]/[`OracleError::DuplicateProxy`]
1355    /// when a discovered feed collides with an existing registration. On
1356    /// error the runtime is unchanged: feeds inserted earlier in the same
1357    /// call are rolled back and the adapter is not retained.
1358    pub async fn register_adapter<A>(
1359        &mut self,
1360        adapter: A,
1361        cache: &mut EvmCache,
1362    ) -> Result<OracleRuntimeMutationReport, OracleError>
1363    where
1364        A: OracleAdapterPlugin + 'static,
1365    {
1366        let adapter = Arc::new(adapter);
1367        let discovery = adapter
1368            .discover(OracleDiscoveryContext {
1369                cache,
1370                now_timestamp: self.tracker.now_timestamp(),
1371            })
1372            .await?;
1373
1374        // Insert discovered feeds live and roll back on failure instead of
1375        // cloning the whole tracker for every call. This is an exact inverse:
1376        // discovery ran before any mutation, `insert_seeded_registration`
1377        // touches only the registration map, the proxy-id index, and the
1378        // snapshot map, and `remove_by_id` removes exactly those entries plus
1379        // pending reconciliations — which a just-inserted proxy cannot have,
1380        // because nothing runs between the inserts and the rollback.
1381        let mut registered_feed_ids: Vec<FeedId> = Vec::new();
1382        let mut registrations = Vec::new();
1383        for feed in &discovery.feeds {
1384            if let Err(error) = self
1385                .tracker
1386                .insert_seeded_registration(feed.registration.clone(), feed.round.clone())
1387            {
1388                for id in registered_feed_ids {
1389                    self.tracker.remove_by_id(id);
1390                }
1391                return Err(error);
1392            }
1393            registered_feed_ids.push(feed.registration.id.clone());
1394            registrations.push(feed.registration.clone());
1395        }
1396
1397        self.push_adapter_state(adapter, registrations);
1398        let mut feed_statuses =
1399            feed_readiness_from_registrations(self.tracker.registrations_iter());
1400        feed_statuses.extend(discovery.skipped.iter().map(skipped_feed_readiness));
1401        Ok(OracleRuntimeMutationReport {
1402            registered_feed_ids,
1403            removed_feed_ids: Vec::new(),
1404            skipped: discovery.skipped,
1405            feed_statuses,
1406            current_handler_ids: self.reactive_handler_ids(),
1407        })
1408    }
1409}
1410
1411#[derive(Clone, Copy, Debug)]
1412enum OracleHandlerRefreshBackfill {
1413    Continuity,
1414    Explicit(SubscriberBackfill),
1415    LiveOnly,
1416}
1417
1418fn runtime_error(error: impl ToString) -> OracleError {
1419    OracleError::Reactive(error.to_string())
1420}
1421
1422fn adapter_not_installed(adapter_id: &str) -> OracleError {
1423    OracleError::Config(crate::error::OracleConfigError::AdapterNotInstalled {
1424        adapter: crate::OracleAdapterId::new(adapter_id.to_string()),
1425    })
1426}
1427
1428fn feed_readiness_from_registrations<'a>(
1429    registrations: impl IntoIterator<Item = &'a FeedRegistration>,
1430) -> Vec<OracleFeedReadinessReport> {
1431    registrations
1432        .into_iter()
1433        .map(|registration| OracleFeedReadinessReport {
1434            id: Some(registration.id.clone()),
1435            proxy: registration.proxy,
1436            status: registration.status,
1437            reason: None,
1438        })
1439        .collect()
1440}
1441
1442fn skipped_feed_readiness(skipped: &OracleAdapterFeedSkip) -> OracleFeedReadinessReport {
1443    OracleFeedReadinessReport {
1444        id: skipped.feed.id(),
1445        proxy: skipped.proxy,
1446        status: OracleFeedStatus::Unsupported,
1447        reason: Some(skipped.reason.to_string()),
1448    }
1449}
1450
1451fn storage_warmup_feed_readiness(
1452    registrations: &[&FeedRegistration],
1453    failed_slots: &[OracleStorageWarmupFailure],
1454) -> Vec<OracleFeedReadinessReport> {
1455    if failed_slots.is_empty() {
1456        return feed_readiness_from_registrations(registrations.iter().copied());
1457    }
1458
1459    let failed_addresses = failed_slots
1460        .iter()
1461        .map(|failure| failure.address)
1462        .collect::<BTreeSet<_>>();
1463    registrations
1464        .iter()
1465        .map(|registration| {
1466            let failed = feed_storage_addresses(registration)
1467                .into_iter()
1468                .any(|address| failed_addresses.contains(&address));
1469            OracleFeedReadinessReport {
1470                id: Some(registration.id.clone()),
1471                proxy: registration.proxy,
1472                status: if failed {
1473                    OracleFeedStatus::Degraded
1474                } else {
1475                    registration.status
1476                },
1477                reason: failed
1478                    .then(|| "oracle storage warmup failed for one or more feed slots".to_string()),
1479            }
1480        })
1481        .collect()
1482}
1483
1484fn feed_storage_addresses(registration: &FeedRegistration) -> Vec<Address> {
1485    let mut addresses = vec![registration.proxy];
1486    addresses.extend(registration.current_aggregator);
1487    addresses.extend(
1488        registration
1489            .source
1490            .event_aggregators(registration.current_aggregator),
1491    );
1492    addresses.sort_unstable();
1493    addresses.dedup();
1494    addresses
1495}
1496
1497fn unregister_handler_ids<S>(
1498    engine: &mut ReactiveEngine<S, Ethereum>,
1499    handler_ids: &[HandlerId],
1500) -> Vec<HandlerId>
1501where
1502    S: InterestOwnerSubscriber<Ethereum>,
1503{
1504    let mut removed = Vec::new();
1505    for id in handler_ids {
1506        if engine.unregister_handler(id).is_some() {
1507            removed.push(id.clone());
1508        }
1509    }
1510    removed
1511}
1512
1513fn cache_runtime_skip_error(skipped: &OracleAdapterFeedSkip) -> OracleError {
1514    OracleError::FeedSkipped(Box::new(crate::OracleFeedSkip::from_adapter_skip(skipped)))
1515}
1516
1517fn prewarm_oracle_storage(
1518    cache: &mut EvmCache,
1519    storage_sync: &OracleStorageSync,
1520    registrations: &[&crate::FeedRegistration],
1521    mode: OracleStorageWarmupMode,
1522) -> OracleStorageWarmupReport {
1523    let slots = storage_sync.warm_slots_for_registrations(registrations.iter().copied());
1524    let discovery_calls = oracle_cold_start_discovery_calls(registrations);
1525    let requested_slots = slots.len();
1526    if slots.is_empty() && discovery_calls.is_empty() {
1527        return OracleStorageWarmupReport {
1528            mode,
1529            feed_statuses: feed_readiness_from_registrations(registrations.iter().copied()),
1530            ..OracleStorageWarmupReport::default()
1531        };
1532    }
1533
1534    match mode {
1535        OracleStorageWarmupMode::PrewarmSlots => {
1536            let report = cache.prewarm_slots(&slots);
1537            let failed_slots = report
1538                .failed
1539                .into_iter()
1540                .map(|(address, slot, error)| OracleStorageWarmupFailure {
1541                    address,
1542                    slot,
1543                    reason: error.to_string(),
1544                })
1545                .collect::<Vec<_>>();
1546            let feed_statuses = storage_warmup_feed_readiness(registrations, &failed_slots);
1547            OracleStorageWarmupReport {
1548                mode,
1549                feed_statuses,
1550                requested_slots,
1551                loaded_slots: report.loaded,
1552                failed_slots,
1553                discovery_calls: 0,
1554                cold_start: None,
1555            }
1556        }
1557        OracleStorageWarmupMode::ColdStart => {
1558            let mut planner =
1559                OracleSlotColdStartPlanner::new(slots.clone(), discovery_calls.clone());
1560            match cache.run_cold_start(&mut planner, ColdStartConfig::default()) {
1561                Ok(report) => {
1562                    let failed = report.failed_slots;
1563                    let cold_start = OracleColdStartWarmupReport::from(report);
1564                    OracleStorageWarmupReport {
1565                        mode,
1566                        feed_statuses: feed_readiness_from_registrations(
1567                            registrations.iter().copied(),
1568                        ),
1569                        requested_slots,
1570                        loaded_slots: requested_slots.saturating_sub(failed),
1571                        failed_slots: Vec::new(),
1572                        discovery_calls: cold_start.discover_calls,
1573                        cold_start: Some(cold_start),
1574                    }
1575                }
1576                Err(error) => {
1577                    let failed_slots = slots
1578                        .into_iter()
1579                        .map(|(address, slot)| OracleStorageWarmupFailure {
1580                            address,
1581                            slot,
1582                            reason: error.to_string(),
1583                        })
1584                        .collect::<Vec<_>>();
1585                    let feed_statuses = storage_warmup_feed_readiness(registrations, &failed_slots);
1586                    OracleStorageWarmupReport {
1587                        mode,
1588                        feed_statuses,
1589                        requested_slots,
1590                        loaded_slots: 0,
1591                        failed_slots,
1592                        discovery_calls: discovery_calls.len(),
1593                        cold_start: None,
1594                    }
1595                }
1596            }
1597        }
1598    }
1599}
1600
1601struct OracleSlotColdStartPlanner {
1602    slots: Vec<(Address, U256)>,
1603    slot_set: BTreeSet<(Address, U256)>,
1604    discover: Vec<ColdStartCall>,
1605    verified_discovered_slots: bool,
1606}
1607
1608impl OracleSlotColdStartPlanner {
1609    fn new(slots: Vec<(Address, U256)>, discover: Vec<ColdStartCall>) -> Self {
1610        let slot_set = slots.iter().copied().collect();
1611        Self {
1612            slots,
1613            slot_set,
1614            discover,
1615            verified_discovered_slots: false,
1616        }
1617    }
1618}
1619
1620impl ColdStartPlanner for OracleSlotColdStartPlanner {
1621    fn initial_plan(&mut self, _state: &dyn StateView) -> ColdStartPlan {
1622        ColdStartPlan {
1623            verify: self.slots.clone(),
1624            discover: self.discover.clone(),
1625            ..Default::default()
1626        }
1627    }
1628
1629    fn on_results(&mut self, results: &ColdStartResults, _state: &dyn StateView) -> ColdStartStep {
1630        if self.verified_discovered_slots {
1631            return ColdStartStep::Done;
1632        }
1633        self.verified_discovered_slots = true;
1634
1635        let mut slots = results
1636            .discovered
1637            .iter()
1638            .flat_map(|discovered| discovered.access.slots.iter().copied())
1639            .filter(|slot| !self.slot_set.contains(slot))
1640            .collect::<Vec<_>>();
1641        slots.sort_unstable();
1642        slots.dedup();
1643        if slots.is_empty() {
1644            return ColdStartStep::Done;
1645        }
1646
1647        ColdStartStep::Continue(ColdStartPlan {
1648            verify: slots,
1649            ..Default::default()
1650        })
1651    }
1652}
1653
1654impl From<evm_fork_cache::ColdStartRunReport> for OracleColdStartWarmupReport {
1655    fn from(report: evm_fork_cache::ColdStartRunReport) -> Self {
1656        Self {
1657            rounds: report.rounds,
1658            verified_slots: report.verified_slots,
1659            changed_slots: report.changed_slots,
1660            failed_slots: report.failed_slots,
1661            discovered_slots: report.discovered_slots,
1662            discovered_accounts: report.discovered_accounts,
1663            discover_calls: report
1664                .per_round
1665                .iter()
1666                .map(|round| round.discover_calls)
1667                .sum(),
1668        }
1669    }
1670}
1671
1672fn oracle_cold_start_discovery_calls(
1673    registrations: &[&crate::FeedRegistration],
1674) -> Vec<ColdStartCall> {
1675    registrations
1676        .iter()
1677        .map(|registration| {
1678            let read_proxy = registration.source.read_proxy(registration.proxy);
1679            ColdStartCall {
1680                from: Address::ZERO,
1681                to: read_proxy,
1682                calldata: Bytes::from(latestRoundDataCall {}.abi_encode()),
1683                restrict_to: Some(vec![read_proxy]),
1684            }
1685        })
1686        .collect()
1687}
1688
1689#[cfg(test)]
1690mod tests {
1691    use super::*;
1692    use evm_fork_cache::{ColdStartCallResult, StorageAccessList};
1693    use revm::context::result::{ExecutionResult, Output, SuccessReason};
1694
1695    #[derive(Default)]
1696    struct EmptyStateView;
1697
1698    impl StateView for EmptyStateView {
1699        fn storage(&self, _address: Address, _slot: U256) -> Option<U256> {
1700            None
1701        }
1702    }
1703
1704    #[test]
1705    fn cold_start_planner_verifies_discovered_slots_in_second_round() {
1706        let declared = (Address::repeat_byte(0x11), U256::from(1_u64));
1707        let discovered = (Address::repeat_byte(0x22), U256::from(2_u64));
1708        let mut planner = OracleSlotColdStartPlanner::new(vec![declared], Vec::new());
1709        let initial = planner.initial_plan(&EmptyStateView);
1710        assert_eq!(initial.verify, vec![declared]);
1711
1712        let mut access = StorageAccessList::default();
1713        access.slots.insert(discovered);
1714        access.slots.insert(declared);
1715        let results = ColdStartResults {
1716            discovered: vec![ColdStartCallResult {
1717                result: ExecutionResult::Success {
1718                    reason: SuccessReason::Return,
1719                    gas_used: 0,
1720                    gas_refunded: 0,
1721                    logs: Vec::new(),
1722                    output: Output::Call(Bytes::new()),
1723                },
1724                access,
1725            }],
1726            ..Default::default()
1727        };
1728
1729        let ColdStartStep::Continue(next) = planner.on_results(&results, &EmptyStateView) else {
1730            panic!("discovered slots should schedule a follow-up verify round");
1731        };
1732        assert_eq!(next.verify, vec![discovered]);
1733        assert!(next.discover.is_empty());
1734
1735        assert!(matches!(
1736            planner.on_results(&ColdStartResults::default(), &EmptyStateView),
1737            ColdStartStep::Done
1738        ));
1739    }
1740}
evm_oracle_state/runtime.rs

evm_oracle_state/
runtime.rs
