revm_database_interface/
async_db.rs

1//! Async database interface.
2use crate::{DBErrorMarker, Database, DatabaseCommit, DatabaseRef};
3use core::{
4    convert::Infallible,
5    future::Future,
6    marker::PhantomData,
7    pin::Pin,
8    ptr::NonNull,
9    task::{Context, Poll},
10};
11use corosensei::{stack::DefaultStack, Coroutine, CoroutineResult, Yielder};
12use primitives::{Address, AddressMap, StorageKey, StorageValue, B256};
13use state::{Account, AccountId, AccountInfo, Bytecode};
14use std::{cell::Cell, fmt, io};
15use tokio::{
16    runtime::{Handle, Runtime},
17    task,
18};
19
20type Resume = AsyncResult<NonNull<Context<'static>>>;
21type Yield = ();
22type Complete<R> = AsyncResult<R>;
23type DatabaseFiber<R> = Coroutine<Resume, Yield, Complete<R>, DefaultStack>;
24
25const DEFAULT_STACK_SIZE: usize = 1024 * 1024;
26
27/// Reusable async EVM fiber stack storage.
28#[derive(Default)]
29pub struct FiberStack {
30    stack: Option<DefaultStack>,
31}
32
33impl Clone for FiberStack {
34    #[inline]
35    fn clone(&self) -> Self {
36        Self::default()
37    }
38}
39
40impl fmt::Debug for FiberStack {
41    #[inline]
42    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43        f.debug_struct("FiberStack").finish_non_exhaustive()
44    }
45}
46
47impl FiberStack {
48    #[inline]
49    fn take_or_new(&mut self) -> AsyncResult<DefaultStack> {
50        match self.stack.take() {
51            Some(stack) => Ok(stack),
52            None => DefaultStack::new(DEFAULT_STACK_SIZE).map_err(AsyncError::Io),
53        }
54    }
55
56    #[inline]
57    fn put(&mut self, stack: DefaultStack) {
58        debug_assert!(self.stack.is_none());
59        self.stack = Some(stack);
60    }
61}
62
63thread_local! {
64    static CURRENT: Cell<Option<NonNull<CurrentFiber>>> = const { Cell::new(None) };
65}
66
67/// Result type used by async database execution helpers.
68pub type AsyncResult<T, E = Infallible> = Result<T, AsyncError<E>>;
69
70/// Error returned by async database execution helpers.
71#[derive(Debug, thiserror::Error)]
72#[non_exhaustive]
73pub enum AsyncError<E = Infallible> {
74    /// The async EVM fiber was cancelled before execution completed.
75    #[error("async EVM execution was cancelled")]
76    Cancelled,
77    /// An async host operation was called outside an async EVM fiber.
78    #[error("async host operation requires EVM async fiber execution")]
79    NotOnFiber,
80    /// Blocking async I/O was requested outside a supported Tokio runtime.
81    #[error("async host operation requires a Tokio multi-thread runtime")]
82    Runtime,
83    /// Async fiber stack setup failed.
84    #[error(transparent)]
85    Io(io::Error),
86    /// The wrapped operation returned an error.
87    #[error(transparent)]
88    Inner(#[from] E),
89}
90
91impl AsyncError {
92    fn with_inner_error<E>(self) -> AsyncError<E> {
93        match self {
94            Self::Cancelled => AsyncError::Cancelled,
95            Self::NotOnFiber => AsyncError::NotOnFiber,
96            Self::Runtime => AsyncError::Runtime,
97            Self::Io(error) => AsyncError::Io(error),
98            Self::Inner(error) => match error {},
99        }
100    }
101}
102
103impl<E: DBErrorMarker> DBErrorMarker for AsyncError<E> {
104    #[inline]
105    fn is_fatal(&self) -> bool {
106        match self {
107            Self::Inner(error) => error.is_fatal(),
108            _ => true,
109        }
110    }
111}
112
113struct CurrentFiber {
114    suspend: NonNull<Yielder<Resume, Yield>>,
115    future_cx: NonNull<Context<'static>>,
116    cancelled: bool,
117}
118
119impl CurrentFiber {
120    #[inline]
121    fn context(&mut self) -> &mut Context<'_> {
122        unsafe { restore_context_lifetime(self.future_cx.as_mut()) }
123    }
124
125    #[inline]
126    fn suspend(&mut self) -> AsyncResult<()> {
127        match unsafe { self.suspend.as_ref() }.suspend(()) {
128            Ok(cx) => {
129                self.future_cx = cx;
130                Ok(())
131            }
132            Err(error) => {
133                self.cancelled = true;
134                Err(error)
135            }
136        }
137    }
138
139    #[inline]
140    const fn is_cancelled(&self) -> bool {
141        self.cancelled
142    }
143}
144
145struct ResetCurrentFiber(Option<NonNull<CurrentFiber>>);
146
147impl Drop for ResetCurrentFiber {
148    fn drop(&mut self) {
149        CURRENT.set(self.0);
150    }
151}
152
153/// Runs `func` on a native fiber and awaits its completion.
154///
155/// Synchronous code running inside `func` may call [`block_on_current`] to wait for async host
156/// operations without blocking the executor thread.
157#[cfg(test)]
158pub(crate) fn on_fiber_result<'a, R, E>(
159    func: impl FnOnce() -> Result<R, E> + 'a,
160) -> impl Future<Output = AsyncResult<R, E>> + Send + 'a
161where
162    R: Send + 'a,
163    E: Send + 'a,
164{
165    OnFiber::new(func)
166}
167
168/// Runs `func` on a native fiber backed by a reusable EVM stack slot.
169///
170/// # Safety
171///
172/// `stack` must point to valid stack storage for the lifetime of the returned future. That storage
173/// must not be accessed by anything else until the returned future is dropped.
174pub unsafe fn on_fiber_result_with_stack<'a, R, E>(
175    stack: NonNull<FiberStack>,
176    func: impl FnOnce() -> Result<R, E> + 'a,
177) -> impl Future<Output = AsyncResult<R, E>> + Send + 'a
178where
179    R: Send + 'a,
180    E: Send + 'a,
181{
182    OnFiber::with_stack(stack, func)
183}
184
185#[cfg(test)]
186pub(crate) fn on_fiber<'a, R>(
187    func: impl FnOnce() -> R + 'a,
188) -> impl Future<Output = AsyncResult<R>> + Send + 'a
189where
190    R: Send + 'a,
191{
192    on_fiber_result(move || Ok::<_, Infallible>(func()))
193}
194
195/// Runs `func` on a native fiber backed by a reusable EVM stack slot.
196///
197/// # Safety
198///
199/// See [`on_fiber_result_with_stack`].
200pub unsafe fn on_fiber_with_stack<'a, R>(
201    stack: NonNull<FiberStack>,
202    func: impl FnOnce() -> R + 'a,
203) -> impl Future<Output = AsyncResult<R>> + Send + 'a
204where
205    R: Send + 'a,
206{
207    unsafe { on_fiber_result_with_stack(stack, move || Ok::<_, Infallible>(func())) }
208}
209
210enum OnFiber<'a, R, E> {
211    Running(FiberFuture<'a, Result<R, E>>),
212    Error(Option<AsyncError>),
213    Done,
214}
215
216impl<'a, R, E> OnFiber<'a, R, E> {
217    #[cfg(test)]
218    fn new(func: impl FnOnce() -> Result<R, E> + 'a) -> Self {
219        Self::new_inner(None, func)
220    }
221
222    fn with_stack(stack: NonNull<FiberStack>, func: impl FnOnce() -> Result<R, E> + 'a) -> Self {
223        Self::new_inner(Some(stack), func)
224    }
225
226    fn new_inner(
227        stack: Option<NonNull<FiberStack>>,
228        func: impl FnOnce() -> Result<R, E> + 'a,
229    ) -> Self {
230        match FiberFuture::new(stack, func) {
231            Ok(fiber) => Self::Running(fiber),
232            Err(error) => Self::Error(Some(error)),
233        }
234    }
235}
236
237impl<R, E> Future for OnFiber<'_, R, E> {
238    type Output = AsyncResult<R, E>;
239
240    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
241        let this = self.get_mut();
242        match this {
243            Self::Running(fiber) => match Pin::new(fiber).poll(cx) {
244                Poll::Ready(Ok(Ok(value))) => {
245                    *this = Self::Done;
246                    Poll::Ready(Ok(value))
247                }
248                Poll::Ready(Ok(Err(error))) => {
249                    *this = Self::Done;
250                    Poll::Ready(Err(AsyncError::Inner(error)))
251                }
252                Poll::Ready(Err(error)) => {
253                    *this = Self::Done;
254                    Poll::Ready(Err(error.with_inner_error()))
255                }
256                Poll::Pending => Poll::Pending,
257            },
258            Self::Error(error) => {
259                let error = error
260                    .take()
261                    .expect("async EVM fiber error already returned");
262                Poll::Ready(Err(error.with_inner_error()))
263            }
264            Self::Done => panic!("async EVM fiber polled after completion"),
265        }
266    }
267}
268
269struct FiberFuture<'a, R> {
270    fiber: Option<DatabaseFiber<R>>,
271    stack: Option<NonNull<FiberStack>>,
272    _marker: PhantomData<&'a ()>,
273}
274
275// SAFETY: The future may move between polls, but the coroutine stack itself is heap allocated and
276// is only resumed through `poll` with a fresh task context. Values that can remain on the coroutine
277// stack across suspension are required to be `Send` by the blocking boundary.
278unsafe impl<R: Send> Send for FiberFuture<'_, R> {}
279
280impl<'a, R> FiberFuture<'a, R> {
281    fn new(
282        mut stack: Option<NonNull<FiberStack>>,
283        func: impl FnOnce() -> R + 'a,
284    ) -> AsyncResult<Self> {
285        let fiber_stack = match &mut stack {
286            Some(stack) => unsafe { stack.as_mut() }.take_or_new()?,
287            None => DefaultStack::new(DEFAULT_STACK_SIZE).map_err(AsyncError::Io)?,
288        };
289        let body = move |suspend: &Yielder<Resume, Yield>, resume| {
290            let future_cx = resume?;
291            let mut current = CurrentFiber {
292                suspend: NonNull::from(suspend),
293                future_cx,
294                cancelled: false,
295            };
296            let current = NonNull::from(&mut current);
297            let previous = CURRENT.replace(Some(current));
298            let _reset = ResetCurrentFiber(previous);
299            Ok(func())
300        };
301        // SAFETY: The coroutine is stored inside `FiberFuture<'a, R>`, which is tied to the
302        // borrowed state lifetime and dropped before those borrows can expire.
303        let fiber = unsafe { Coroutine::with_stack_unchecked(fiber_stack, body) };
304        Ok(Self {
305            fiber: Some(fiber),
306            stack,
307            _marker: PhantomData,
308        })
309    }
310
311    fn recycle_stack(&mut self) {
312        let Some(fiber) = self.fiber.take() else {
313            return;
314        };
315        debug_assert!(fiber.done());
316        let stack = fiber.into_stack();
317        if let Some(mut slot) = self.stack {
318            unsafe { slot.as_mut() }.put(stack);
319        }
320    }
321}
322
323impl<R> Future for FiberFuture<'_, R> {
324    type Output = AsyncResult<R>;
325
326    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
327        let this = self.get_mut();
328        let cx = NonNull::from(unsafe { change_context_lifetime(cx) });
329        let fiber = this
330            .fiber
331            .as_mut()
332            .expect("async EVM fiber polled after completion");
333        match fiber.resume(Ok(cx)) {
334            CoroutineResult::Return(result) => {
335                this.recycle_stack();
336                Poll::Ready(result)
337            }
338            CoroutineResult::Yield(()) => Poll::Pending,
339        }
340    }
341}
342
343impl<R> Drop for FiberFuture<'_, R> {
344    fn drop(&mut self) {
345        let Some(fiber) = self.fiber.as_mut() else {
346            return;
347        };
348        if fiber.done() {
349            self.recycle_stack();
350        } else if matches!(
351            fiber.resume(Err(AsyncError::Cancelled)),
352            CoroutineResult::Yield(())
353        ) {
354            // SAFETY: Cancellation already gave the coroutine a chance to return normally. If it
355            // yields again, the stack is no longer useful to this future.
356            unsafe { fiber.force_reset() };
357        } else {
358            self.recycle_stack();
359        }
360    }
361}
362
363/// Polls `future` to completion from inside an async EVM fiber.
364///
365/// If `future` returns `Poll::Pending`, the current EVM fiber is suspended and the outer async EVM
366/// future returns `Poll::Pending`. When the executor wakes and polls the outer future again, the
367/// EVM fiber resumes and continues polling `future`.
368///
369/// # Errors
370///
371/// Returns [`AsyncError::NotOnFiber`] if called outside async EVM execution, or
372/// [`AsyncError::Cancelled`] if the outer async EVM execution was dropped.
373pub fn block_on_current<F: Future>(future: F) -> AsyncResult<F::Output> {
374    let mut future = core::pin::pin!(future);
375    loop {
376        match with_current(|current| {
377            if current.is_cancelled() {
378                return Err(AsyncError::Cancelled);
379            }
380            let poll = future.as_mut().poll(current.context());
381            if poll.is_pending() {
382                current.suspend()?;
383            }
384            Ok(poll)
385        })? {
386            Poll::Ready(value) => return Ok(value),
387            Poll::Pending => {}
388        }
389    }
390}
391
392fn current_tokio_handle() -> Option<Handle> {
393    match Handle::try_current() {
394        Ok(handle) => match handle.runtime_flavor() {
395            tokio::runtime::RuntimeFlavor::CurrentThread => None,
396            _ => Some(handle),
397        },
398        Err(_) => None,
399    }
400}
401
402fn block_on_handle<F>(handle: &Handle, future: F) -> F::Output
403where
404    F: Future + Send,
405    F::Output: Send,
406{
407    let should_use_block_in_place = Handle::try_current()
408        .ok()
409        .map(|current| {
410            !matches!(
411                current.runtime_flavor(),
412                tokio::runtime::RuntimeFlavor::CurrentThread
413            )
414        })
415        .unwrap_or(false);
416
417    if should_use_block_in_place {
418        task::block_in_place(move || handle.block_on(future))
419    } else {
420        handle.block_on(future)
421    }
422}
423
424fn block_on_runtime<F>(runtime: Option<&Handle>, future: F) -> AsyncResult<F::Output>
425where
426    F: Future + Send,
427    F::Output: Send,
428{
429    if CURRENT.get().is_some() {
430        return block_on_current(future);
431    }
432
433    if let Some(runtime) = runtime {
434        return Ok(block_on_handle(runtime, future));
435    }
436
437    Err(AsyncError::Runtime)
438}
439
440fn block_on_runtime_result<F, T, E>(runtime: Option<&Handle>, future: F) -> AsyncResult<T, E>
441where
442    F: Future<Output = Result<T, E>> + Send,
443    T: Send,
444    E: Send,
445{
446    match block_on_runtime(runtime, future).map_err(AsyncError::with_inner_error)? {
447        Ok(value) => Ok(value),
448        Err(error) => Err(AsyncError::Inner(error)),
449    }
450}
451
452fn with_current<R>(f: impl FnOnce(&mut CurrentFiber) -> AsyncResult<R>) -> AsyncResult<R> {
453    let mut current = CURRENT.get().ok_or(AsyncError::NotOnFiber)?;
454    f(unsafe { current.as_mut() })
455}
456
457unsafe fn change_context_lifetime<'a>(cx: &'a mut Context<'_>) -> &'a mut Context<'static> {
458    unsafe { core::mem::transmute::<&'a mut Context<'_>, &'a mut Context<'static>>(cx) }
459}
460
461unsafe fn restore_context_lifetime<'a>(cx: &'a mut Context<'static>) -> &'a mut Context<'a> {
462    unsafe { core::mem::transmute::<&'a mut Context<'static>, &'a mut Context<'a>>(cx) }
463}
464
465/// The async EVM database interface.
466///
467/// Contains the same methods as [Database], but it returns [Future] type instead.
468///
469/// Use [AsyncDb] to provide [Database] implementation for a type that only implements this trait.
470pub trait DatabaseAsync {
471    /// The database error type.
472    type Error: DBErrorMarker;
473
474    /// Gets basic account information.
475    fn basic_async(
476        &mut self,
477        address: Address,
478    ) -> impl Future<Output = Result<Option<AccountInfo>, Self::Error>> + Send;
479
480    /// Gets account code by its hash.
481    fn code_by_hash_async(
482        &mut self,
483        code_hash: B256,
484    ) -> impl Future<Output = Result<Bytecode, Self::Error>> + Send;
485
486    /// Gets storage value of address at index.
487    fn storage_async(
488        &mut self,
489        address: Address,
490        index: StorageKey,
491    ) -> impl Future<Output = Result<StorageValue, Self::Error>> + Send;
492
493    /// Gets storage value of account by its id.
494    ///
495    /// Default implementation is to call [`DatabaseAsync::storage_async`] method.
496    #[inline]
497    fn storage_by_account_id_async(
498        &mut self,
499        address: Address,
500        account_id: AccountId,
501        storage_key: StorageKey,
502    ) -> impl Future<Output = Result<StorageValue, Self::Error>> + Send {
503        let _ = account_id;
504        self.storage_async(address, storage_key)
505    }
506
507    /// Gets block hash by block number.
508    fn block_hash_async(
509        &mut self,
510        number: u64,
511    ) -> impl Future<Output = Result<B256, Self::Error>> + Send;
512}
513
514/// The async EVM database interface.
515///
516/// Contains the same methods as [DatabaseRef], but it returns [Future] type instead.
517///
518/// Use [AsyncDb] to provide [DatabaseRef] implementation for a type that only implements this trait.
519pub trait DatabaseAsyncRef {
520    /// The database error type.
521    type Error: DBErrorMarker;
522
523    /// Gets basic account information.
524    fn basic_async_ref(
525        &self,
526        address: Address,
527    ) -> impl Future<Output = Result<Option<AccountInfo>, Self::Error>> + Send;
528
529    /// Gets account code by its hash.
530    fn code_by_hash_async_ref(
531        &self,
532        code_hash: B256,
533    ) -> impl Future<Output = Result<Bytecode, Self::Error>> + Send;
534
535    /// Gets storage value of address at index.
536    fn storage_async_ref(
537        &self,
538        address: Address,
539        index: StorageKey,
540    ) -> impl Future<Output = Result<StorageValue, Self::Error>> + Send;
541
542    /// Gets storage value of account by its id.
543    ///
544    /// Default implementation is to call [`DatabaseAsyncRef::storage_async_ref`] method.
545    #[inline]
546    fn storage_by_account_id_async_ref(
547        &self,
548        address: Address,
549        account_id: AccountId,
550        storage_key: StorageKey,
551    ) -> impl Future<Output = Result<StorageValue, Self::Error>> + Send {
552        let _ = account_id;
553        self.storage_async_ref(address, storage_key)
554    }
555
556    /// Gets block hash by block number.
557    fn block_hash_async_ref(
558        &self,
559        number: u64,
560    ) -> impl Future<Output = Result<B256, Self::Error>> + Send;
561}
562
563/// Adapter that exposes an async database through the synchronous [`Database`] interface.
564#[derive(Debug)]
565pub struct AsyncDb<T> {
566    db: T,
567    rt: Option<HandleOrRuntime>,
568}
569
570impl<T> AsyncDb<T> {
571    /// Creates a new async database adapter.
572    ///
573    /// This captures the current Tokio runtime handle when one is available.
574    #[inline]
575    pub fn new(db: T) -> Self {
576        Self {
577            db,
578            rt: current_tokio_handle().map(HandleOrRuntime::Handle),
579        }
580    }
581
582    /// Creates a new async database adapter using the current Tokio runtime handle.
583    ///
584    /// Returns `None` if no Tokio runtime is available or the current runtime is current-threaded.
585    #[inline]
586    pub fn blocking(db: T) -> Option<Self> {
587        Some(Self {
588            db,
589            rt: Some(HandleOrRuntime::Handle(current_tokio_handle()?)),
590        })
591    }
592
593    /// Creates a new async database adapter with a Tokio runtime.
594    #[inline]
595    pub const fn with_runtime(db: T, runtime: Runtime) -> Self {
596        Self {
597            db,
598            rt: Some(HandleOrRuntime::Runtime(runtime)),
599        }
600    }
601
602    /// Creates a new async database adapter with a Tokio runtime handle.
603    #[inline]
604    pub const fn with_handle(db: T, handle: Handle) -> Self {
605        Self {
606            db,
607            rt: Some(HandleOrRuntime::Handle(handle)),
608        }
609    }
610
611    /// Returns the wrapped database.
612    #[inline]
613    pub const fn inner(&self) -> &T {
614        &self.db
615    }
616
617    /// Returns the wrapped database mutably.
618    #[inline]
619    pub const fn inner_mut(&mut self) -> &mut T {
620        &mut self.db
621    }
622
623    /// Consumes the adapter and returns the wrapped database.
624    #[inline]
625    pub fn into_inner(self) -> T {
626        self.db
627    }
628}
629
630impl<T: DatabaseAsync> Database for AsyncDb<T> {
631    type Error = AsyncError<T::Error>;
632
633    #[inline]
634    fn basic(&mut self, address: Address) -> Result<Option<AccountInfo>, Self::Error> {
635        let Self { db, rt } = self;
636        block_on_runtime_result(
637            rt.as_ref().map(HandleOrRuntime::handle),
638            db.basic_async(address),
639        )
640    }
641
642    #[inline]
643    fn code_by_hash(&mut self, code_hash: B256) -> Result<Bytecode, Self::Error> {
644        let Self { db, rt } = self;
645        block_on_runtime_result(
646            rt.as_ref().map(HandleOrRuntime::handle),
647            db.code_by_hash_async(code_hash),
648        )
649    }
650
651    #[inline]
652    fn storage(
653        &mut self,
654        address: Address,
655        index: StorageKey,
656    ) -> Result<StorageValue, Self::Error> {
657        let Self { db, rt } = self;
658        block_on_runtime_result(
659            rt.as_ref().map(HandleOrRuntime::handle),
660            db.storage_async(address, index),
661        )
662    }
663
664    #[inline]
665    fn storage_by_account_id(
666        &mut self,
667        address: Address,
668        account_id: AccountId,
669        storage_key: StorageKey,
670    ) -> Result<StorageValue, Self::Error> {
671        let Self { db, rt } = self;
672        block_on_runtime_result(
673            rt.as_ref().map(HandleOrRuntime::handle),
674            db.storage_by_account_id_async(address, account_id, storage_key),
675        )
676    }
677
678    #[inline]
679    fn block_hash(&mut self, number: u64) -> Result<B256, Self::Error> {
680        let Self { db, rt } = self;
681        block_on_runtime_result(
682            rt.as_ref().map(HandleOrRuntime::handle),
683            db.block_hash_async(number),
684        )
685    }
686}
687
688impl<T: DatabaseAsyncRef> DatabaseRef for AsyncDb<T> {
689    type Error = AsyncError<T::Error>;
690
691    #[inline]
692    fn basic_ref(&self, address: Address) -> Result<Option<AccountInfo>, Self::Error> {
693        block_on_runtime_result(
694            self.rt.as_ref().map(HandleOrRuntime::handle),
695            self.db.basic_async_ref(address),
696        )
697    }
698
699    #[inline]
700    fn code_by_hash_ref(&self, code_hash: B256) -> Result<Bytecode, Self::Error> {
701        block_on_runtime_result(
702            self.rt.as_ref().map(HandleOrRuntime::handle),
703            self.db.code_by_hash_async_ref(code_hash),
704        )
705    }
706
707    #[inline]
708    fn storage_ref(
709        &self,
710        address: Address,
711        index: StorageKey,
712    ) -> Result<StorageValue, Self::Error> {
713        block_on_runtime_result(
714            self.rt.as_ref().map(HandleOrRuntime::handle),
715            self.db.storage_async_ref(address, index),
716        )
717    }
718
719    #[inline]
720    fn storage_by_account_id_ref(
721        &self,
722        address: Address,
723        account_id: AccountId,
724        storage_key: StorageKey,
725    ) -> Result<StorageValue, Self::Error> {
726        block_on_runtime_result(
727            self.rt.as_ref().map(HandleOrRuntime::handle),
728            self.db
729                .storage_by_account_id_async_ref(address, account_id, storage_key),
730        )
731    }
732
733    #[inline]
734    fn block_hash_ref(&self, number: u64) -> Result<B256, Self::Error> {
735        block_on_runtime_result(
736            self.rt.as_ref().map(HandleOrRuntime::handle),
737            self.db.block_hash_async_ref(number),
738        )
739    }
740}
741
742impl<T: DatabaseAsync + DatabaseCommit> DatabaseCommit for AsyncDb<T> {
743    #[inline]
744    fn commit(&mut self, changes: AddressMap<Account>) {
745        self.db.commit(changes);
746    }
747
748    #[inline]
749    fn commit_iter(&mut self, changes: &mut dyn Iterator<Item = (Address, Account)>) {
750        self.db.commit_iter(changes);
751    }
752}
753
754/// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] to provide a [`Database`] implementation.
755#[derive(Debug)]
756pub struct WrapDatabaseAsync<T>(AsyncDb<T>);
757
758impl<T> WrapDatabaseAsync<T> {
759    /// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] instance.
760    ///
761    /// Returns `None` if no tokio runtime is available or if the current runtime is a current-thread runtime.
762    #[inline]
763    pub fn new(db: T) -> Option<Self> {
764        AsyncDb::blocking(db).map(Self)
765    }
766
767    /// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] instance, with a runtime.
768    ///
769    /// Refer to [tokio::runtime::Builder] on how to create a runtime if you are in synchronous world.
770    ///
771    /// If you are already using something like [tokio::main], call [`WrapDatabaseAsync::new`] instead.
772    #[inline]
773    pub const fn with_runtime(db: T, runtime: Runtime) -> Self {
774        Self(AsyncDb::with_runtime(db, runtime))
775    }
776
777    /// Wraps a [DatabaseAsync] or [DatabaseAsyncRef] instance, with a runtime handle.
778    ///
779    /// This generally allows you to pass any valid runtime handle, refer to [tokio::runtime::Handle] on how
780    /// to obtain a handle.
781    ///
782    /// If you are already in asynchronous world, like [tokio::main], use [`WrapDatabaseAsync::new`] instead.
783    #[inline]
784    pub const fn with_handle(db: T, handle: Handle) -> Self {
785        Self(AsyncDb::with_handle(db, handle))
786    }
787
788    /// Returns the wrapped database.
789    #[inline]
790    pub const fn inner(&self) -> &T {
791        self.0.inner()
792    }
793
794    /// Returns the wrapped database mutably.
795    #[inline]
796    pub const fn inner_mut(&mut self) -> &mut T {
797        self.0.inner_mut()
798    }
799
800    /// Consumes the adapter and returns the wrapped database.
801    #[inline]
802    pub fn into_inner(self) -> T {
803        self.0.into_inner()
804    }
805}
806
807impl<T: DatabaseAsync> Database for WrapDatabaseAsync<T> {
808    type Error = AsyncError<T::Error>;
809
810    #[inline]
811    fn basic(&mut self, address: Address) -> Result<Option<AccountInfo>, Self::Error> {
812        self.0.basic(address)
813    }
814
815    #[inline]
816    fn code_by_hash(&mut self, code_hash: B256) -> Result<Bytecode, Self::Error> {
817        self.0.code_by_hash(code_hash)
818    }
819
820    #[inline]
821    fn storage(
822        &mut self,
823        address: Address,
824        index: StorageKey,
825    ) -> Result<StorageValue, Self::Error> {
826        self.0.storage(address, index)
827    }
828
829    #[inline]
830    fn storage_by_account_id(
831        &mut self,
832        address: Address,
833        account_id: AccountId,
834        storage_key: StorageKey,
835    ) -> Result<StorageValue, Self::Error> {
836        self.0
837            .storage_by_account_id(address, account_id, storage_key)
838    }
839
840    #[inline]
841    fn block_hash(&mut self, number: u64) -> Result<B256, Self::Error> {
842        self.0.block_hash(number)
843    }
844}
845
846impl<T: DatabaseAsyncRef> DatabaseRef for WrapDatabaseAsync<T> {
847    type Error = AsyncError<T::Error>;
848
849    #[inline]
850    fn basic_ref(&self, address: Address) -> Result<Option<AccountInfo>, Self::Error> {
851        self.0.basic_ref(address)
852    }
853
854    #[inline]
855    fn code_by_hash_ref(&self, code_hash: B256) -> Result<Bytecode, Self::Error> {
856        self.0.code_by_hash_ref(code_hash)
857    }
858
859    #[inline]
860    fn storage_ref(
861        &self,
862        address: Address,
863        index: StorageKey,
864    ) -> Result<StorageValue, Self::Error> {
865        self.0.storage_ref(address, index)
866    }
867
868    #[inline]
869    fn storage_by_account_id_ref(
870        &self,
871        address: Address,
872        account_id: AccountId,
873        storage_key: StorageKey,
874    ) -> Result<StorageValue, Self::Error> {
875        self.0
876            .storage_by_account_id_ref(address, account_id, storage_key)
877    }
878
879    #[inline]
880    fn block_hash_ref(&self, number: u64) -> Result<B256, Self::Error> {
881        self.0.block_hash_ref(number)
882    }
883}
884
885impl<T: DatabaseAsync + DatabaseCommit> DatabaseCommit for WrapDatabaseAsync<T> {
886    #[inline]
887    fn commit(&mut self, changes: AddressMap<Account>) {
888        self.0.commit(changes);
889    }
890
891    #[inline]
892    fn commit_iter(&mut self, changes: &mut dyn Iterator<Item = (Address, Account)>) {
893        self.0.commit_iter(changes);
894    }
895}
896
897// Hold a tokio runtime handle or full runtime.
898#[derive(Debug)]
899enum HandleOrRuntime {
900    Handle(Handle),
901    Runtime(Runtime),
902}
903
904impl HandleOrRuntime {
905    #[inline]
906    fn handle(&self) -> &Handle {
907        match self {
908            Self::Handle(handle) => handle,
909            Self::Runtime(runtime) => runtime.handle(),
910        }
911    }
912}
913
914#[cfg(test)]
915mod tests {
916    use super::{block_on_current, on_fiber, AsyncDb, AsyncError, DatabaseAsync};
917    use crate::Database;
918    use core::{convert::Infallible, fmt, future::Future, pin::Pin, task::Poll};
919    use primitives::{Address, StorageKey, StorageValue, B256};
920    use state::{AccountInfo, Bytecode};
921    use std::task::{Context, Waker};
922
923    #[test]
924    fn block_on_requires_fiber() {
925        assert!(matches!(
926            block_on_current(core::future::ready(())),
927            Err(AsyncError::NotOnFiber)
928        ));
929    }
930
931    #[test]
932    fn fiber_suspends_and_resumes_pending_future() {
933        let mut state = 1;
934        let mut future = core::pin::pin!(on_fiber(|| {
935            state += block_on_current(PendingOnce { pending: true }).unwrap();
936            state
937        }));
938        let waker = Waker::noop();
939        let mut cx = Context::from_waker(waker);
940
941        assert!(matches!(future.as_mut().poll(&mut cx), Poll::Pending));
942        assert!(matches!(future.as_mut().poll(&mut cx), Poll::Ready(Ok(3))));
943    }
944
945    #[test]
946    fn fiber_reuses_stack_slot() {
947        let mut stack = super::FiberStack::default();
948        let stack_ptr = core::ptr::NonNull::from(&mut stack);
949
950        poll_ready(unsafe {
951            super::on_fiber_result_with_stack(stack_ptr, || Ok::<_, Infallible>(1))
952        })
953        .unwrap();
954        assert!(stack.stack.is_some());
955        poll_ready(unsafe {
956            super::on_fiber_result_with_stack(stack_ptr, || Ok::<_, Infallible>(2))
957        })
958        .unwrap();
959        assert!(stack.stack.is_some());
960    }
961
962    #[test]
963    fn async_database_adapts_to_database() {
964        let mut db = AsyncDb::new(TestDb);
965
966        let value = poll_ready(on_fiber(|| {
967            Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap()
968        }))
969        .unwrap();
970
971        assert_eq!(value, StorageValue::from(9));
972    }
973
974    #[test]
975    fn async_database_suspends_until_ready() {
976        let mut db = AsyncDb::new(PendingDb { pending: true });
977        let mut future = core::pin::pin!(on_fiber(|| {
978            Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap()
979        }));
980        let waker = Waker::noop();
981        let mut cx = Context::from_waker(waker);
982
983        assert!(matches!(future.as_mut().poll(&mut cx), Poll::Pending));
984        assert!(
985            matches!(future.as_mut().poll(&mut cx), Poll::Ready(Ok(value)) if value == StorageValue::from(9))
986        );
987    }
988
989    #[test]
990    fn async_database_returns_database_error() {
991        let mut db = AsyncDb::new(FailingDb);
992        let result = poll_ready(on_fiber(|| {
993            Database::storage(&mut db, Address::ZERO, StorageKey::from(7))
994        }));
995
996        assert!(matches!(result, Ok(Err(AsyncError::Inner(TestError)))));
997    }
998
999    #[test]
1000    fn synchronous_database_blocks_with_current_tokio_runtime() {
1001        let runtime = tokio::runtime::Runtime::new().unwrap();
1002        let _guard = runtime.enter();
1003        let mut db = AsyncDb::new(TokioDb);
1004
1005        let value = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap();
1006
1007        assert_eq!(value, StorageValue::from(9));
1008    }
1009
1010    #[test]
1011    fn blocking_constructor_uses_current_tokio_runtime() {
1012        let runtime = tokio::runtime::Runtime::new().unwrap();
1013        let _guard = runtime.enter();
1014        let mut db = AsyncDb::blocking(TokioDb).unwrap();
1015
1016        let value = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap();
1017
1018        assert_eq!(value, StorageValue::from(9));
1019    }
1020
1021    #[test]
1022    fn synchronous_database_blocks_with_stored_tokio_handle() {
1023        let runtime = tokio::runtime::Runtime::new().unwrap();
1024        let mut db = AsyncDb::with_handle(TokioDb, runtime.handle().clone());
1025
1026        let value = Database::storage(&mut db, Address::ZERO, StorageKey::from(7)).unwrap();
1027
1028        assert_eq!(value, StorageValue::from(9));
1029    }
1030
1031    #[test]
1032    fn synchronous_database_requires_runtime_handle() {
1033        let mut db = AsyncDb::new(TestDb);
1034
1035        let result = Database::storage(&mut db, Address::ZERO, StorageKey::from(7));
1036
1037        assert!(matches!(result, Err(AsyncError::Runtime)));
1038    }
1039
1040    #[test]
1041    fn dropping_fiber_cancels_blocked_future() {
1042        let mut saw_cancel = false;
1043        {
1044            let mut future = core::pin::pin!(on_fiber(|| {
1045                saw_cancel = matches!(block_on_current(PendingForever), Err(AsyncError::Cancelled));
1046            }));
1047            let waker = Waker::noop();
1048            let mut cx = Context::from_waker(waker);
1049
1050            assert!(matches!(future.as_mut().poll(&mut cx), Poll::Pending));
1051        }
1052        assert!(saw_cancel);
1053    }
1054
1055    fn poll_ready<F: Future + Send>(future: F) -> F::Output {
1056        let mut future = core::pin::pin!(future);
1057        let waker = Waker::noop();
1058        let mut cx = Context::from_waker(waker);
1059
1060        match future.as_mut().poll(&mut cx) {
1061            Poll::Ready(value) => value,
1062            Poll::Pending => panic!("future unexpectedly pending"),
1063        }
1064    }
1065
1066    struct PendingOnce {
1067        pending: bool,
1068    }
1069
1070    impl Future for PendingOnce {
1071        type Output = i32;
1072
1073        fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
1074            if self.pending {
1075                self.pending = false;
1076                Poll::Pending
1077            } else {
1078                Poll::Ready(2)
1079            }
1080        }
1081    }
1082
1083    struct PendingForever;
1084
1085    impl Future for PendingForever {
1086        type Output = ();
1087
1088        fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
1089            Poll::Pending
1090        }
1091    }
1092
1093    struct TestDb;
1094
1095    impl DatabaseAsync for TestDb {
1096        type Error = Infallible;
1097
1098        async fn basic_async(
1099            &mut self,
1100            _address: Address,
1101        ) -> Result<Option<AccountInfo>, Self::Error> {
1102            Ok(None)
1103        }
1104
1105        async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result<Bytecode, Self::Error> {
1106            Ok(Bytecode::default())
1107        }
1108
1109        async fn storage_async(
1110            &mut self,
1111            _address: Address,
1112            _index: StorageKey,
1113        ) -> Result<StorageValue, Self::Error> {
1114            Ok(StorageValue::from(9))
1115        }
1116
1117        async fn block_hash_async(&mut self, _number: u64) -> Result<B256, Self::Error> {
1118            Ok(B256::ZERO)
1119        }
1120    }
1121
1122    struct PendingDb {
1123        pending: bool,
1124    }
1125
1126    impl DatabaseAsync for PendingDb {
1127        type Error = Infallible;
1128
1129        async fn basic_async(
1130            &mut self,
1131            _address: Address,
1132        ) -> Result<Option<AccountInfo>, Self::Error> {
1133            Ok(None)
1134        }
1135
1136        async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result<Bytecode, Self::Error> {
1137            Ok(Bytecode::default())
1138        }
1139
1140        async fn storage_async(
1141            &mut self,
1142            _address: Address,
1143            _index: StorageKey,
1144        ) -> Result<StorageValue, Self::Error> {
1145            PendingStorage {
1146                pending: &mut self.pending,
1147            }
1148            .await;
1149            Ok(StorageValue::from(9))
1150        }
1151
1152        async fn block_hash_async(&mut self, _number: u64) -> Result<B256, Self::Error> {
1153            Ok(B256::ZERO)
1154        }
1155    }
1156
1157    struct PendingStorage<'a> {
1158        pending: &'a mut bool,
1159    }
1160
1161    impl Future for PendingStorage<'_> {
1162        type Output = ();
1163
1164        fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
1165            if *self.pending {
1166                *self.pending = false;
1167                Poll::Pending
1168            } else {
1169                Poll::Ready(())
1170            }
1171        }
1172    }
1173
1174    struct FailingDb;
1175
1176    impl DatabaseAsync for FailingDb {
1177        type Error = TestError;
1178
1179        async fn basic_async(
1180            &mut self,
1181            _address: Address,
1182        ) -> Result<Option<AccountInfo>, Self::Error> {
1183            Ok(None)
1184        }
1185
1186        async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result<Bytecode, Self::Error> {
1187            Ok(Bytecode::default())
1188        }
1189
1190        async fn storage_async(
1191            &mut self,
1192            _address: Address,
1193            _index: StorageKey,
1194        ) -> Result<StorageValue, Self::Error> {
1195            Err(TestError)
1196        }
1197
1198        async fn block_hash_async(&mut self, _number: u64) -> Result<B256, Self::Error> {
1199            Ok(B256::ZERO)
1200        }
1201    }
1202
1203    struct TokioDb;
1204
1205    impl DatabaseAsync for TokioDb {
1206        type Error = Infallible;
1207
1208        async fn basic_async(
1209            &mut self,
1210            _address: Address,
1211        ) -> Result<Option<AccountInfo>, Self::Error> {
1212            tokio::task::yield_now().await;
1213            Ok(None)
1214        }
1215
1216        async fn code_by_hash_async(&mut self, _code_hash: B256) -> Result<Bytecode, Self::Error> {
1217            tokio::task::yield_now().await;
1218            Ok(Bytecode::default())
1219        }
1220
1221        async fn storage_async(
1222            &mut self,
1223            _address: Address,
1224            _index: StorageKey,
1225        ) -> Result<StorageValue, Self::Error> {
1226            tokio::task::yield_now().await;
1227            Ok(StorageValue::from(9))
1228        }
1229
1230        async fn block_hash_async(&mut self, _number: u64) -> Result<B256, Self::Error> {
1231            tokio::task::yield_now().await;
1232            Ok(B256::ZERO)
1233        }
1234    }
1235
1236    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
1237    struct TestError;
1238
1239    impl fmt::Display for TestError {
1240        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1241            f.write_str("storage read failed")
1242        }
1243    }
1244
1245    impl core::error::Error for TestError {}
1246
1247    impl crate::DBErrorMarker for TestError {}
1248}
revm_database_interface/async_db.rs

revm_database_interface/
async_db.rs
