signet_sim/env/
shared.rs

1use crate::{
2    cache::StateSource, env::RollupEnv, outcome::SimulatedItem, AcctInfo, HostEnv, SimCache, SimDb,
3    SimEnv,
4};
5use alloy::primitives::Address;
6use core::fmt;
7use std::{ops::Deref, sync::Arc};
8use tokio::{select, sync::watch};
9use tracing::{debug, debug_span, instrument, trace, warn, Span};
10use trevm::{
11    db::TryCachingDb,
12    helpers::Ctx,
13    revm::{database::Cache, inspector::NoOpInspector, DatabaseRef, Inspector},
14};
15
16/// Composite async source that overlays the sim env's committed cache
17/// on top of a fallback [`StateSource`].
18///
19/// Accounts whose state was modified by prior sim rounds (present in
20/// the cache) are returned directly, avoiding async I/O. Accounts not
21/// in the cache fall through to the asynchronous source.
22struct CachedAsyncSource<'a, S> {
23    cache: &'a Cache,
24    fallback: &'a S,
25}
26
27impl<S: StateSource> StateSource for CachedAsyncSource<'_, S> {
28    type Error = S::Error;
29
30    #[instrument(level = "trace", skip_all, fields(%address, source = tracing::field::Empty))]
31    async fn account_details(&self, address: &Address) -> Result<AcctInfo, Self::Error> {
32        if let Some(acct) = self.cache.accounts.get(address) {
33            Span::current().record("source", "cache_hit");
34            return Ok(AcctInfo {
35                nonce: acct.info.nonce,
36                balance: acct.info.balance,
37                has_code: acct.info.code_hash() != trevm::revm::primitives::KECCAK_EMPTY,
38            });
39        }
40        Span::current().record("source", "fallback");
41        self.fallback.account_details(address).await
42    }
43}
44
45/// A simulation environment.
46///
47/// Contains enough information to run a simulation.
48pub struct SharedSimEnv<RuDb, HostDb, RuInsp = NoOpInspector, HostInsp = NoOpInspector> {
49    inner: Arc<SimEnv<RuDb, HostDb, RuInsp, HostInsp>>,
50}
51
52impl<RuDb, HostDb, RuInsp, HostInsp> fmt::Debug for SharedSimEnv<RuDb, HostDb, RuInsp, HostInsp> {
53    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
54        f.debug_struct("SharedSimEnv")
55            .field("finish_by", &self.inner.finish_by())
56            .field("concurrency_limit", &self.inner.concurrency_limit())
57            .finish_non_exhaustive()
58    }
59}
60
61impl<RuDb, HostDb, RuInsp, HostInsp> Deref for SharedSimEnv<RuDb, HostDb, RuInsp, HostInsp> {
62    type Target = SimEnv<RuDb, HostDb, RuInsp, HostInsp>;
63
64    fn deref(&self) -> &Self::Target {
65        &self.inner
66    }
67}
68
69impl<RuDb, HostDb, RuInsp, HostInsp> From<SimEnv<RuDb, HostDb, RuInsp, HostInsp>>
70    for SharedSimEnv<RuDb, HostDb, RuInsp, HostInsp>
71where
72    RuDb: DatabaseRef + Send + Sync + 'static,
73    RuInsp: Inspector<Ctx<SimDb<RuDb>>> + Default + Sync + 'static,
74    HostDb: DatabaseRef + Send + Sync + 'static,
75    HostInsp: Inspector<Ctx<SimDb<HostDb>>> + Default + Sync + 'static,
76{
77    fn from(inner: SimEnv<RuDb, HostDb, RuInsp, HostInsp>) -> Self {
78        Self { inner: Arc::new(inner) }
79    }
80}
81
82impl<RuDb, HostDb, RuInsp, HostInsp> SharedSimEnv<RuDb, HostDb, RuInsp, HostInsp>
83where
84    RuDb: DatabaseRef + Send + Sync + 'static,
85    RuInsp: Inspector<Ctx<SimDb<RuDb>>> + Default + Sync + 'static,
86    HostDb: DatabaseRef + Send + Sync + 'static,
87    HostInsp: Inspector<Ctx<SimDb<HostDb>>> + Default + Sync + 'static,
88{
89    /// Creates a new `SimEnv` instance.
90    pub fn new(
91        rollup: RollupEnv<RuDb, RuInsp>,
92        host: HostEnv<HostDb, HostInsp>,
93        finish_by: tokio::time::Instant,
94        concurrency_limit: usize,
95        sim_items: SimCache,
96    ) -> Self {
97        SimEnv::new(rollup, host, finish_by, concurrency_limit, sim_items).into()
98    }
99
100    /// Get a reference the simulation cache used by this builder.
101    pub fn sim_items(&self) -> &SimCache {
102        self.inner.sim_items()
103    }
104
105    /// Get a reference to the rollup environment.
106    pub fn rollup_env(&self) -> &RollupEnv<RuDb, RuInsp> {
107        self.inner.rollup_env()
108    }
109
110    /// Get a mutable reference to the rollup environment.
111    pub fn rollup_env_mut(&mut self) -> &mut RollupEnv<RuDb, RuInsp> {
112        Arc::get_mut(&mut self.inner).expect("sims dropped already").rollup_mut()
113    }
114
115    /// Get a reference to the host environment.
116    pub fn host_env(&self) -> &HostEnv<HostDb, HostInsp> {
117        self.inner.host_env()
118    }
119
120    /// Get a mutable reference to the host environment.
121    pub fn host_env_mut(&mut self) -> &mut HostEnv<HostDb, HostInsp> {
122        Arc::get_mut(&mut self.inner).expect("sims dropped already").host_mut()
123    }
124
125    /// Run a simulation round, returning the best item.
126    ///
127    /// Preflight validity checks (nonce/balance) are performed asynchronously
128    /// using the provided [`StateSource`]s. This avoids the tokio I/O
129    /// driver starvation deadlock that occurs when sync `DatabaseRef` calls
130    /// go through `block_in_place` + `Handle::block_on`.
131    pub async fn sim_round<AS, AH>(
132        &mut self,
133        max_gas: u64,
134        max_host_gas: u64,
135        async_ru_source: &AS,
136        async_host_source: &AH,
137    ) -> Option<SimulatedItem>
138    where
139        AS: StateSource,
140        AH: StateSource,
141    {
142        let span = debug_span!(
143            "sim_round",
144            max_gas,
145            max_host_gas,
146            items_to_simulate = tracing::field::Empty,
147            items_simulated_ok = tracing::field::Empty,
148            items_simulated_err = tracing::field::Empty,
149        )
150        .or_current();
151
152        // Overlay the sim env's committed cache so that accounts touched
153        // by prior rounds (e.g. nonce bumps) are visible to the preflight
154        // validity check without requiring async I/O.
155        let ru_source = CachedAsyncSource {
156            cache: self.inner.rollup_env().db().cache(),
157            fallback: async_ru_source,
158        };
159        let host_source = CachedAsyncSource {
160            cache: self.inner.host_env().db().cache(),
161            fallback: async_host_source,
162        };
163
164        let active_sim = match self
165            .inner
166            .sim_items()
167            .read_best_valid(self.inner.concurrency_limit(), &ru_source, &host_source)
168            .await
169        {
170            Ok(items) => items,
171            Err(error) => {
172                warn!(%error, "preflight validity check failed");
173                return None;
174            }
175        };
176
177        span.record("items_to_simulate", active_sim.len());
178
179        if active_sim.is_empty() {
180            return None;
181        }
182
183        // These will be moved into the blocking task.
184        let scope_span = span.clone();
185        let this = self.inner.clone();
186        let (best_tx, mut best_watcher) = watch::channel(None);
187
188        // Spawn a blocking task to run the simulations.
189        let sim_task = tokio::task::spawn_blocking(move || {
190            scope_span.in_scope(|| this.sim_round(max_gas, max_host_gas, best_tx, active_sim))
191        });
192
193        // Either simulation is done, or we time out
194        let sim_counts = select! {
195            _ = tokio::time::sleep_until(self.finish_by()) => {
196                span.in_scope(|| trace!("Sim round timed out"));
197                None
198            },
199            result = sim_task => {
200                span.in_scope(|| trace!("Sim round done"));
201                result.ok()
202            },
203        };
204
205        if let Some(counts) = sim_counts {
206            span.record("items_simulated_ok", counts.ok);
207            span.record("items_simulated_err", counts.err);
208        }
209
210        let _guard = span.entered();
211
212        // Check what the current best outcome is.
213        let best = best_watcher.borrow_and_update();
214        trace!(score = %best.as_ref().map(|candidate| candidate.score).unwrap_or_default(), "Read outcome from channel");
215        let outcome = best.as_ref()?;
216
217        // Remove the item from the cache.
218        let item = self.sim_items().remove(outcome.cache_rank)?;
219
220        // We can expect here as all of our simulations are done and cleaned up.
221        let inner = Arc::get_mut(&mut self.inner).expect("sims dropped already");
222
223        // Accept the cache from the simulation.
224        inner.rollup_mut().accept_cache_ref(&outcome.rollup_cache).ok()?;
225        // Accept the host cache from the simulation.
226        inner.host_mut().accept_cache_ref(&outcome.host_cache).ok()?;
227        // Accept the aggregate fills and orders.
228        inner
229            .rollup_mut()
230            .accept_aggregates(&outcome.bundle_fills, &outcome.bundle_orders)
231            .expect("checked during simulation");
232
233        debug!(
234            score = %outcome.score,
235            gas_used = outcome.gas_used,
236            host_gas_used = outcome.host_gas_used,
237            identifier = %item.identifier(),
238            "Selected simulated item",
239        );
240
241        Some(SimulatedItem {
242            gas_used: outcome.gas_used,
243            host_gas_used: outcome.host_gas_used,
244            score: outcome.score,
245            item,
246        })
247    }
248}
signet_sim/env/shared.rs

signet_sim/env/
shared.rs
