Skip to main content

reth_db/implementation/mdbx/
mod.rs

1//! Module that interacts with MDBX.
2
3use crate::{
4    lockfile::StorageLock,
5    metrics::DatabaseEnvMetrics,
6    tables::{self, Tables},
7    utils::default_page_size,
8    DatabaseError, TableSet,
9};
10use eyre::Context;
11use metrics::{gauge, Label};
12use reth_db_api::{
13    cursor::{DbCursorRO, DbCursorRW},
14    database::Database,
15    database_metrics::DatabaseMetrics,
16    models::ClientVersion,
17    transaction::{DbTx, DbTxMut},
18};
19use reth_libmdbx::{
20    ffi, DatabaseFlags, Environment, EnvironmentFlags, Geometry, HandleSlowReadersReturnCode,
21    MaxReadTransactionDuration, Mode, PageSize, SyncMode, RO, RW,
22};
23use reth_storage_errors::db::LogLevel;
24use reth_tracing::tracing::error;
25use rustc_hash::FxHashMap;
26use std::{
27    ops::{Deref, Range},
28    path::{Path, PathBuf},
29    sync::Arc,
30    time::{SystemTime, UNIX_EPOCH},
31};
32use tx::Tx;
33
34pub mod cursor;
35pub mod tx;
36
37mod utils;
38#[cfg(target_os = "linux")]
39mod value_prefetch;
40
41/// 1 KB in bytes
42pub const KILOBYTE: usize = 1024;
43/// 1 MB in bytes
44pub const MEGABYTE: usize = KILOBYTE * 1024;
45/// 1 GB in bytes
46pub const GIGABYTE: usize = MEGABYTE * 1024;
47/// 1 TB in bytes
48pub const TERABYTE: usize = GIGABYTE * 1024;
49
50/// MDBX allows up to 32767 readers (`MDBX_READERS_LIMIT`), but we limit it to slightly below that
51const DEFAULT_MAX_READERS: u64 = 32_000;
52
53/// Space that a read-only transaction can occupy until the warning is emitted.
54/// See [`reth_libmdbx::EnvironmentBuilder::set_handle_slow_readers`] for more information.
55const MAX_SAFE_READER_SPACE: usize = 10 * GIGABYTE;
56
57/// Environment used when opening a MDBX environment. RO/RW.
58#[derive(Clone, Copy, Debug, Eq, PartialEq)]
59pub enum DatabaseEnvKind {
60    /// Read-only MDBX environment.
61    RO,
62    /// Read-write MDBX environment.
63    RW,
64}
65
66impl DatabaseEnvKind {
67    /// Returns `true` if the environment is read-write.
68    pub const fn is_rw(&self) -> bool {
69        matches!(self, Self::RW)
70    }
71}
72
73/// Arguments for database initialization.
74#[derive(Clone, Debug)]
75pub struct DatabaseArguments {
76    /// Client version that accesses the database.
77    client_version: ClientVersion,
78    /// Database geometry settings.
79    geometry: Geometry<Range<usize>>,
80    /// Database log level. If [None], the default value is used.
81    log_level: Option<LogLevel>,
82    /// Maximum duration of a read transaction. If [None], the default value is used.
83    max_read_transaction_duration: Option<MaxReadTransactionDuration>,
84    /// Open environment in exclusive/monopolistic mode. If [None], the default value is used.
85    ///
86    /// This can be used as a replacement for `MDB_NOLOCK`, which don't supported by MDBX. In this
87    /// way, you can get the minimal overhead, but with the correct multi-process and multi-thread
88    /// locking.
89    ///
90    /// If `true` = open environment in exclusive/monopolistic mode or return `MDBX_BUSY` if
91    /// environment already used by other process. The main feature of the exclusive mode is the
92    /// ability to open the environment placed on a network share.
93    ///
94    /// If `false` = open environment in cooperative mode, i.e. for multi-process
95    /// access/interaction/cooperation. The main requirements of the cooperative mode are:
96    /// - Data files MUST be placed in the LOCAL file system, but NOT on a network share.
97    /// - Environment MUST be opened only by LOCAL processes, but NOT over a network.
98    /// - OS kernel (i.e. file system and memory mapping implementation) and all processes that
99    ///   open the given environment MUST be running in the physically single RAM with
100    ///   cache-coherency. The only exception for cache-consistency requirement is Linux on MIPS
101    ///   architecture, but this case has not been tested for a long time).
102    ///
103    /// This flag affects only at environment opening but can't be changed after.
104    exclusive: Option<bool>,
105    /// MDBX allows up to 32767 readers (`MDBX_READERS_LIMIT`). This arg is to configure the max
106    /// readers.
107    max_readers: Option<u64>,
108    /// Defines the synchronization strategy used by the MDBX database when writing data to disk.
109    ///
110    /// This determines how aggressively MDBX ensures data durability versus prioritizing
111    /// performance. The available modes are:
112    ///
113    /// - [`SyncMode::Durable`]: Ensures all transactions are fully flushed to disk before they are
114    ///   considered committed.   This provides the highest level of durability and crash safety
115    ///   but may have a performance cost.
116    /// - [`SyncMode::SafeNoSync`]: Skips certain fsync operations to improve write performance.
117    ///   This mode still maintains database integrity but may lose the most recent transactions if
118    ///   the system crashes unexpectedly.
119    ///
120    /// Choose `Durable` if consistency and crash safety are critical (e.g., production
121    /// environments). Choose `SafeNoSync` if performance is more important and occasional data
122    /// loss is acceptable (e.g., testing or ephemeral data).
123    sync_mode: SyncMode,
124}
125
126impl Default for DatabaseArguments {
127    fn default() -> Self {
128        Self::new(ClientVersion::default())
129    }
130}
131
132impl DatabaseArguments {
133    /// Create new database arguments with given client version.
134    pub fn new(client_version: ClientVersion) -> Self {
135        Self {
136            client_version,
137            geometry: Geometry {
138                size: Some(0..(8 * TERABYTE)),
139                growth_step: Some(4 * GIGABYTE as isize),
140                shrink_threshold: Some(0),
141                page_size: Some(PageSize::Set(default_page_size())),
142            },
143            log_level: None,
144            max_read_transaction_duration: None,
145            exclusive: None,
146            max_readers: None,
147            sync_mode: SyncMode::Durable,
148        }
149    }
150
151    /// Create database arguments suitable for testing.
152    ///
153    /// Uses a small geometry (64MB max, 4MB growth) to avoid exhausting the system's
154    /// virtual memory map limit (`vm.max_map_count`) when many test databases are open
155    /// concurrently.
156    pub fn test() -> Self {
157        Self {
158            geometry: Geometry {
159                size: Some(0..(64 * MEGABYTE)),
160                growth_step: Some(4 * MEGABYTE as isize),
161                shrink_threshold: Some(0),
162                page_size: Some(PageSize::Set(default_page_size())),
163            },
164            max_read_transaction_duration: Some(MaxReadTransactionDuration::Unbounded),
165            ..Self::new(ClientVersion::default())
166        }
167    }
168
169    /// Sets the upper size limit of the db environment, the maximum database size in bytes.
170    pub const fn with_geometry_max_size(mut self, max_size: Option<usize>) -> Self {
171        if let Some(max_size) = max_size {
172            self.geometry.size = Some(0..max_size);
173        }
174        self
175    }
176
177    /// Sets the database page size value.
178    pub const fn with_geometry_page_size(mut self, page_size: Option<usize>) -> Self {
179        if let Some(size) = page_size {
180            self.geometry.page_size = Some(reth_libmdbx::PageSize::Set(size));
181        }
182
183        self
184    }
185
186    /// Sets the database sync mode.
187    pub const fn with_sync_mode(mut self, sync_mode: Option<SyncMode>) -> Self {
188        if let Some(sync_mode) = sync_mode {
189            self.sync_mode = sync_mode;
190        }
191
192        self
193    }
194
195    /// Configures the database growth step in bytes.
196    pub const fn with_growth_step(mut self, growth_step: Option<usize>) -> Self {
197        if let Some(growth_step) = growth_step {
198            self.geometry.growth_step = Some(growth_step as isize);
199        }
200        self
201    }
202
203    /// Set the log level.
204    pub const fn with_log_level(mut self, log_level: Option<LogLevel>) -> Self {
205        self.log_level = log_level;
206        self
207    }
208
209    /// Set the maximum duration of a read transaction.
210    pub const fn max_read_transaction_duration(
211        &mut self,
212        max_read_transaction_duration: Option<MaxReadTransactionDuration>,
213    ) {
214        self.max_read_transaction_duration = max_read_transaction_duration;
215    }
216
217    /// Set the maximum duration of a read transaction.
218    pub const fn with_max_read_transaction_duration(
219        mut self,
220        max_read_transaction_duration: Option<MaxReadTransactionDuration>,
221    ) -> Self {
222        self.max_read_transaction_duration(max_read_transaction_duration);
223        self
224    }
225
226    /// Set the mdbx exclusive flag.
227    pub const fn with_exclusive(mut self, exclusive: Option<bool>) -> Self {
228        self.exclusive = exclusive;
229        self
230    }
231
232    /// Set `max_readers` flag.
233    pub const fn with_max_readers(mut self, max_readers: Option<u64>) -> Self {
234        self.max_readers = max_readers;
235        self
236    }
237
238    /// Returns the client version if any.
239    pub const fn client_version(&self) -> &ClientVersion {
240        &self.client_version
241    }
242}
243
244/// Wrapper for the libmdbx environment: [Environment]
245#[derive(Debug, Clone)]
246pub struct DatabaseEnv {
247    /// Libmdbx-sys environment.
248    inner: Environment,
249    /// Path to the database directory.
250    path: PathBuf,
251    /// Opened DBIs for reuse.
252    /// Important: Do not manually close these DBIs, like via `mdbx_dbi_close`.
253    /// More generally, do not dynamically create, re-open, or drop tables at
254    /// runtime. It's better to perform table creation and migration only once
255    /// at startup.
256    dbis: Arc<FxHashMap<&'static str, ffi::MDBX_dbi>>,
257    /// Cache for metric handles. If `None`, metrics are not recorded.
258    metrics: Option<Arc<DatabaseEnvMetrics>>,
259    /// Write lock for when dealing with a read-write environment.
260    _lock_file: Option<StorageLock>,
261}
262
263impl Database for DatabaseEnv {
264    type TX = tx::Tx<RO>;
265    type TXMut = tx::Tx<RW>;
266
267    fn tx(&self) -> Result<Self::TX, DatabaseError> {
268        Tx::new(
269            self.inner.begin_ro_txn().map_err(|e| DatabaseError::InitTx(e.into()))?,
270            self.dbis.clone(),
271            self.metrics.clone(),
272        )
273        .map_err(|e| DatabaseError::InitTx(e.into()))
274    }
275
276    fn tx_mut(&self) -> Result<Self::TXMut, DatabaseError> {
277        Tx::new(
278            self.inner.begin_rw_txn().map_err(|e| DatabaseError::InitTx(e.into()))?,
279            self.dbis.clone(),
280            self.metrics.clone(),
281        )
282        .map_err(|e| DatabaseError::InitTx(e.into()))
283    }
284
285    fn path(&self) -> PathBuf {
286        self.path.clone()
287    }
288
289    fn oldest_reader_txnid(&self) -> Option<u64> {
290        let info = self.inner.info().ok()?;
291        let txnid = info.latter_reader_txnid();
292        if txnid == 0 {
293            None
294        } else {
295            Some(txnid)
296        }
297    }
298
299    fn last_txnid(&self) -> Option<u64> {
300        let info = self.inner.info().ok()?;
301        let txnid = info.last_txnid();
302        if txnid == 0 {
303            None
304        } else {
305            Some(txnid as u64)
306        }
307    }
308}
309
310impl DatabaseMetrics for DatabaseEnv {
311    fn report_metrics(&self) {
312        for (name, value, labels) in self.gauge_metrics() {
313            gauge!(name, labels).set(value);
314        }
315    }
316
317    fn gauge_metrics(&self) -> Vec<(&'static str, f64, Vec<Label>)> {
318        let mut metrics = Vec::new();
319
320        let _ = self
321            .view(|tx| {
322                for table in Tables::ALL.iter().map(Tables::name) {
323                    let table_db =
324                        tx.inner().open_db(Some(table)).wrap_err("Could not open db.")?;
325
326                    let stats = tx
327                        .inner()
328                        .db_stat(table_db.dbi())
329                        .wrap_err(format!("Could not find table: {table}"))?;
330
331                    let page_size = stats.page_size() as usize;
332                    let leaf_pages = stats.leaf_pages();
333                    let branch_pages = stats.branch_pages();
334                    let overflow_pages = stats.overflow_pages();
335                    let num_pages = leaf_pages + branch_pages + overflow_pages;
336                    let table_size = page_size * num_pages;
337                    let entries = stats.entries();
338
339                    metrics.push((
340                        "db.table_size",
341                        table_size as f64,
342                        vec![Label::new("table", table)],
343                    ));
344                    metrics.push((
345                        "db.table_pages",
346                        leaf_pages as f64,
347                        vec![Label::new("table", table), Label::new("type", "leaf")],
348                    ));
349                    metrics.push((
350                        "db.table_pages",
351                        branch_pages as f64,
352                        vec![Label::new("table", table), Label::new("type", "branch")],
353                    ));
354                    metrics.push((
355                        "db.table_pages",
356                        overflow_pages as f64,
357                        vec![Label::new("table", table), Label::new("type", "overflow")],
358                    ));
359                    metrics.push((
360                        "db.table_entries",
361                        entries as f64,
362                        vec![Label::new("table", table)],
363                    ));
364                }
365
366                Ok::<(), eyre::Report>(())
367            })
368            .map_err(|error| error!(%error, "Failed to read db table stats"));
369
370        if let Ok(freelist) =
371            self.freelist().map_err(|error| error!(%error, "Failed to read db.freelist"))
372        {
373            metrics.push(("db.freelist", freelist as f64, vec![]));
374        }
375
376        if let Ok(stat) = self.stat().map_err(|error| error!(%error, "Failed to read db.stat")) {
377            metrics.push(("db.page_size", stat.page_size() as f64, vec![]));
378        }
379
380        metrics.push((
381            "db.timed_out_not_aborted_transactions",
382            self.timed_out_not_aborted_transactions() as f64,
383            vec![],
384        ));
385
386        metrics
387    }
388}
389
390impl DatabaseEnv {
391    /// Opens the database at the specified path with the given `EnvKind`.
392    ///
393    /// It does not create the tables, for that call [`DatabaseEnv::create_tables`].
394    pub fn open(
395        path: &Path,
396        kind: DatabaseEnvKind,
397        args: DatabaseArguments,
398    ) -> Result<Self, DatabaseError> {
399        let _lock_file = if kind.is_rw() {
400            StorageLock::try_acquire(path)
401                .map_err(|err| DatabaseError::Other(err.to_string()))?
402                .into()
403        } else {
404            None
405        };
406
407        let mut inner_env = Environment::builder();
408
409        let mode = match kind {
410            DatabaseEnvKind::RO => Mode::ReadOnly,
411            DatabaseEnvKind::RW => {
412                // enable writemap mode in RW mode
413                inner_env.write_map();
414                Mode::ReadWrite { sync_mode: args.sync_mode }
415            }
416        };
417
418        // Note: We set max dbs to 256 here to allow for custom tables. This needs to be set on
419        // environment creation.
420        debug_assert!(Tables::ALL.len() <= 256, "number of tables exceed max dbs");
421        inner_env.set_max_dbs(256);
422        inner_env.set_geometry(args.geometry);
423
424        fn is_current_process(id: u32) -> bool {
425            #[cfg(unix)]
426            {
427                id == std::os::unix::process::parent_id() || id == std::process::id()
428            }
429
430            #[cfg(not(unix))]
431            {
432                id == std::process::id()
433            }
434        }
435
436        extern "C" fn handle_slow_readers(
437            _env: *const ffi::MDBX_env,
438            _txn: *const ffi::MDBX_txn,
439            process_id: ffi::mdbx_pid_t,
440            thread_id: ffi::mdbx_tid_t,
441            read_txn_id: u64,
442            gap: std::ffi::c_uint,
443            space: usize,
444            retry: std::ffi::c_int,
445        ) -> HandleSlowReadersReturnCode {
446            if space > MAX_SAFE_READER_SPACE {
447                let message = if is_current_process(process_id as u32) {
448                    "Current process has a long-lived database transaction that grows the database file."
449                } else {
450                    "External process has a long-lived database transaction that grows the database file. \
451                     Use shorter-lived read transactions or shut down the node."
452                };
453                reth_tracing::tracing::warn!(
454                    target: "storage::db::mdbx",
455                    ?process_id,
456                    ?thread_id,
457                    ?read_txn_id,
458                    ?gap,
459                    ?space,
460                    ?retry,
461                    "{message}"
462                )
463            }
464
465            reth_libmdbx::HandleSlowReadersReturnCode::ProceedWithoutKillingReader
466        }
467        inner_env.set_handle_slow_readers(handle_slow_readers);
468
469        inner_env.set_flags(EnvironmentFlags {
470            mode,
471            // We disable readahead because it improves performance for linear scans, but
472            // worsens it for random access (which is our access pattern outside of sync)
473            no_rdahead: true,
474            coalesce: true,
475            exclusive: args.exclusive.unwrap_or_default(),
476            ..Default::default()
477        });
478        // Configure more readers
479        inner_env.set_max_readers(args.max_readers.unwrap_or(DEFAULT_MAX_READERS));
480        // This parameter sets the maximum size of the "reclaimed list", and the unit of measurement
481        // is "pages". Reclaimed list is the list of freed pages that's populated during the
482        // lifetime of DB transaction, and through which MDBX searches when it needs to insert new
483        // record with overflow pages. The flow is roughly the following:
484        // 0. We need to insert a record that requires N number of overflow pages (in consecutive
485        //    sequence inside the DB file).
486        // 1. Get some pages from the freelist, put them into the reclaimed list.
487        // 2. Search through the reclaimed list for the sequence of size N.
488        // 3. a. If found, return the sequence.
489        // 3. b. If not found, repeat steps 1-3. If the reclaimed list size is larger than
490        //    the `rp augment limit`, stop the search and allocate new pages at the end of the file:
491        //    https://github.com/paradigmxyz/reth/blob/2a4c78759178f66e30c8976ec5d243b53102fc9a/crates/storage/libmdbx-rs/mdbx-sys/libmdbx/mdbx.c#L11479-L11480.
492        //
493        // Basically, this parameter controls for how long do we search through the freelist before
494        // trying to allocate new pages. Smaller value will make MDBX to fallback to
495        // allocation faster, higher value will force MDBX to search through the freelist
496        // longer until the sequence of pages is found.
497        //
498        // The default value of this parameter is set depending on the DB size. The bigger the
499        // database, the larger is `rp augment limit`.
500        // https://github.com/paradigmxyz/reth/blob/2a4c78759178f66e30c8976ec5d243b53102fc9a/crates/storage/libmdbx-rs/mdbx-sys/libmdbx/mdbx.c#L10018-L10024.
501        //
502        // Previously, MDBX set this value as `256 * 1024` constant. Let's fallback to this,
503        // because we want to prioritize freelist lookup speed over database growth.
504        // https://github.com/paradigmxyz/reth/blob/fa2b9b685ed9787636d962f4366caf34a9186e66/crates/storage/libmdbx-rs/mdbx-sys/libmdbx/mdbx.c#L16017.
505        inner_env.set_rp_augment_limit(256 * 1024);
506
507        if let Some(log_level) = args.log_level {
508            // Levels higher than [LogLevel::Notice] require libmdbx built with `MDBX_DEBUG` option.
509            let is_log_level_available = if cfg!(debug_assertions) {
510                true
511            } else {
512                matches!(
513                    log_level,
514                    LogLevel::Fatal | LogLevel::Error | LogLevel::Warn | LogLevel::Notice
515                )
516            };
517            if is_log_level_available {
518                inner_env.set_log_level(match log_level {
519                    LogLevel::Fatal => 0,
520                    LogLevel::Error => 1,
521                    LogLevel::Warn => 2,
522                    LogLevel::Notice => 3,
523                    LogLevel::Verbose => 4,
524                    LogLevel::Debug => 5,
525                    LogLevel::Trace => 6,
526                    LogLevel::Extra => 7,
527                });
528            } else {
529                return Err(DatabaseError::LogLevelUnavailable(log_level))
530            }
531        }
532
533        if let Some(max_read_transaction_duration) = args.max_read_transaction_duration {
534            inner_env.set_max_read_transaction_duration(max_read_transaction_duration);
535        }
536
537        let env = Self {
538            inner: inner_env.open(path).map_err(|e| DatabaseError::Open(e.into()))?,
539            path: path.to_path_buf(),
540            dbis: Arc::default(),
541            metrics: None,
542            _lock_file,
543        };
544
545        Ok(env)
546    }
547
548    /// Enables metrics on the database.
549    pub fn with_metrics(mut self) -> Self {
550        self.metrics = Some(DatabaseEnvMetrics::new().into());
551        self
552    }
553
554    /// Enables metrics on the database if requested.
555    pub fn with_metrics_if(self, enabled: bool) -> Self {
556        if enabled {
557            self.with_metrics()
558        } else {
559            self
560        }
561    }
562
563    /// Creates all the tables defined in [`Tables`], if necessary.
564    ///
565    /// This keeps tracks of the created table handles and stores them for better efficiency.
566    pub fn create_tables(&mut self) -> Result<(), DatabaseError> {
567        self.create_and_track_tables_for::<Tables>()
568    }
569
570    /// Creates all the tables defined in the given [`TableSet`], if necessary.
571    ///
572    /// This keeps tracks of the created table handles and stores them for better efficiency.
573    pub fn create_and_track_tables_for<TS: TableSet>(&mut self) -> Result<(), DatabaseError> {
574        let handles = self._create_tables::<TS>()?;
575        // Note: This is okay because self has mutable access here and `DatabaseEnv` must be Arc'ed
576        // before it can be shared.
577        let dbis = Arc::make_mut(&mut self.dbis);
578        dbis.extend(handles);
579
580        Ok(())
581    }
582
583    /// Creates all the tables defined in [`Tables`], if necessary.
584    ///
585    /// If this type is unique the created handle for the tables will be updated.
586    ///
587    /// This is recommended to be called during initialization to create and track additional tables
588    /// after the default [`Self::create_tables`] are created.
589    pub fn create_tables_for<TS: TableSet>(self: &mut Arc<Self>) -> Result<(), DatabaseError> {
590        let handles = self._create_tables::<TS>()?;
591        if let Some(db) = Arc::get_mut(self) {
592            // Note: The db is unique and the dbis as well, and they can also be cloned.
593            let dbis = Arc::make_mut(&mut db.dbis);
594            dbis.extend(handles);
595        }
596        Ok(())
597    }
598
599    /// Creates the tables and returns the identifiers of the tables.
600    fn _create_tables<TS: TableSet>(
601        &self,
602    ) -> Result<Vec<(&'static str, ffi::MDBX_dbi)>, DatabaseError> {
603        let mut handles = Vec::new();
604        let tx = self.inner.begin_rw_txn().map_err(|e| DatabaseError::InitTx(e.into()))?;
605
606        for table in TS::tables() {
607            let flags =
608                if table.is_dupsort() { DatabaseFlags::DUP_SORT } else { DatabaseFlags::default() };
609
610            let db = tx
611                .create_db(Some(table.name()), flags)
612                .map_err(|e| DatabaseError::CreateTable(e.into()))?;
613            handles.push((table.name(), db.dbi()));
614        }
615
616        tx.commit().map_err(|e| DatabaseError::Commit(e.into()))?;
617        Ok(handles)
618    }
619
620    /// Drops an orphaned table by name.
621    ///
622    /// This is used to clean up tables that are no longer defined in the schema but may still
623    /// exist on disk from previous versions.
624    ///
625    /// Returns `Ok(true)` if the table existed and was dropped, `Ok(false)` if the table was not
626    /// found.
627    ///
628    /// # Safety
629    /// This permanently deletes the table and all its data. Only use for tables that are
630    /// confirmed to be obsolete.
631    pub fn drop_orphan_table(&self, name: &str) -> Result<bool, DatabaseError> {
632        let tx = self.inner.begin_rw_txn().map_err(|e| DatabaseError::InitTx(e.into()))?;
633
634        match tx.open_db(Some(name)) {
635            Ok(db) => {
636                // SAFETY: We just opened the db handle and will commit immediately after dropping.
637                // No other cursors or handles exist for this table.
638                unsafe {
639                    tx.drop_db(db.dbi()).map_err(|e| DatabaseError::Delete(e.into()))?;
640                }
641                tx.commit().map_err(|e| DatabaseError::Commit(e.into()))?;
642                Ok(true)
643            }
644            Err(reth_libmdbx::Error::NotFound) => Ok(false),
645            Err(e) => Err(DatabaseError::Open(e.into())),
646        }
647    }
648
649    /// Records version that accesses the database with write privileges.
650    pub fn record_client_version(&self, version: ClientVersion) -> Result<(), DatabaseError> {
651        if version.is_empty() {
652            return Ok(())
653        }
654
655        let tx = self.tx_mut()?;
656        let mut version_cursor = tx.cursor_write::<tables::VersionHistory>()?;
657
658        let last_version = version_cursor.last()?.map(|(_, v)| v);
659        if Some(&version) != last_version.as_ref() {
660            version_cursor.upsert(
661                SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(),
662                &version,
663            )?;
664            tx.commit()?;
665        }
666
667        Ok(())
668    }
669}
670
671impl Deref for DatabaseEnv {
672    type Target = Environment;
673
674    fn deref(&self) -> &Self::Target {
675        &self.inner
676    }
677}
678
679#[cfg(test)]
680mod tests {
681
682    use super::*;
683    use crate::{
684        tables::{
685            AccountsHistory, CanonicalHeaders, Headers, PlainAccountState, PlainStorageState,
686        },
687        test_utils::*,
688        AccountChangeSets,
689    };
690    use alloy_consensus::Header;
691    use alloy_primitives::{address, Address, B256, U256};
692    use reth_db_api::{
693        cursor::{DbDupCursorRO, DbDupCursorRW, ReverseWalker, Walker},
694        models::{AccountBeforeTx, IntegerList, ShardedKey},
695        table::{Encode, Table},
696    };
697    use reth_libmdbx::Error;
698    use reth_primitives_traits::{Account, StorageEntry};
699    use reth_storage_errors::db::{DatabaseWriteError, DatabaseWriteOperation};
700    use tempfile::TempDir;
701
702    /// Create database for testing. Returns the `TempDir` to prevent cleanup until test ends.
703    fn create_test_db(kind: DatabaseEnvKind) -> (TempDir, DatabaseEnv) {
704        let tempdir = tempfile::TempDir::new().expect(ERROR_TEMPDIR);
705        let env = create_test_db_with_path(kind, tempdir.path());
706        (tempdir, env)
707    }
708
709    /// Create database for testing with specified path
710    fn create_test_db_with_path(kind: DatabaseEnvKind, path: &Path) -> DatabaseEnv {
711        let mut env =
712            DatabaseEnv::open(path, kind, DatabaseArguments::new(ClientVersion::default()))
713                .expect(ERROR_DB_CREATION);
714        env.create_tables().expect(ERROR_TABLE_CREATION);
715        env
716    }
717
718    const ERROR_DB_CREATION: &str = "Not able to create the mdbx file.";
719    const ERROR_PUT: &str = "Not able to insert value into table.";
720    const ERROR_APPEND: &str = "Not able to append the value to the table.";
721    const ERROR_UPSERT: &str = "Not able to upsert the value to the table.";
722    const ERROR_GET: &str = "Not able to get value from table.";
723    const ERROR_DEL: &str = "Not able to delete from table.";
724    const ERROR_COMMIT: &str = "Not able to commit transaction.";
725    const ERROR_RETURN_VALUE: &str = "Mismatching result.";
726    const ERROR_INIT_TX: &str = "Failed to create a MDBX transaction.";
727
728    #[test]
729    fn db_creation() {
730        let _tempdir = create_test_db(DatabaseEnvKind::RW);
731    }
732
733    #[test]
734    fn db_drop_orphan_table() {
735        let tempdir = tempfile::TempDir::new().expect(ERROR_TEMPDIR);
736        let db = create_test_db_with_path(DatabaseEnvKind::RW, tempdir.path());
737
738        // Create an orphan table by manually creating it
739        let orphan_table_name = "OrphanTestTable";
740        {
741            let tx = db.inner.begin_rw_txn().expect(ERROR_INIT_TX);
742            tx.create_db(Some(orphan_table_name), DatabaseFlags::empty())
743                .expect("Failed to create orphan table");
744            tx.commit().expect(ERROR_COMMIT);
745        }
746
747        // Verify the table exists by opening it
748        {
749            let tx = db.inner.begin_ro_txn().expect(ERROR_INIT_TX);
750            assert!(tx.open_db(Some(orphan_table_name)).is_ok(), "Orphan table should exist");
751        }
752
753        // Drop the orphan table
754        let result = db.drop_orphan_table(orphan_table_name);
755        assert!(result.is_ok(), "drop_orphan_table should succeed");
756        assert!(result.unwrap(), "drop_orphan_table should return true for existing table");
757
758        // Verify the table no longer exists
759        {
760            let tx = db.inner.begin_ro_txn().expect(ERROR_INIT_TX);
761            assert!(
762                tx.open_db(Some(orphan_table_name)).is_err(),
763                "Orphan table should no longer exist"
764            );
765        }
766
767        // Dropping a non-existent table should return Ok(false)
768        let result = db.drop_orphan_table("NonExistentTable");
769        assert!(result.is_ok(), "drop_orphan_table should succeed for non-existent table");
770        assert!(!result.unwrap(), "drop_orphan_table should return false for non-existent table");
771    }
772
773    #[test]
774    fn db_manual_put_get() {
775        let (_tempdir, env) = create_test_db(DatabaseEnvKind::RW);
776
777        let value = Header::default();
778        let key = 1u64;
779
780        // PUT
781        let tx = env.tx_mut().expect(ERROR_INIT_TX);
782        tx.put::<Headers>(key, value.clone()).expect(ERROR_PUT);
783        tx.commit().expect(ERROR_COMMIT);
784
785        // GET
786        let tx = env.tx().expect(ERROR_INIT_TX);
787        let result = tx.get::<Headers>(key).expect(ERROR_GET);
788        assert_eq!(result.expect(ERROR_RETURN_VALUE), value);
789        tx.commit().expect(ERROR_COMMIT);
790    }
791
792    #[test]
793    fn db_dup_cursor_delete_first() {
794        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
795        let tx = db.tx_mut().expect(ERROR_INIT_TX);
796
797        let mut dup_cursor = tx.cursor_dup_write::<PlainStorageState>().unwrap();
798
799        let entry_0 = StorageEntry { key: B256::with_last_byte(1), value: U256::ZERO };
800        let entry_1 = StorageEntry { key: B256::with_last_byte(1), value: U256::ONE };
801
802        dup_cursor.upsert(Address::with_last_byte(1), &entry_0).expect(ERROR_UPSERT);
803        dup_cursor.upsert(Address::with_last_byte(1), &entry_1).expect(ERROR_UPSERT);
804
805        assert_eq!(
806            dup_cursor.walk(None).unwrap().collect::<Result<Vec<_>, _>>().unwrap(),
807            vec![(Address::with_last_byte(1), entry_0), (Address::with_last_byte(1), entry_1),]
808        );
809
810        let mut walker = dup_cursor.walk(None).unwrap();
811        walker.delete_current().expect(ERROR_DEL);
812
813        assert_eq!(walker.next().unwrap().unwrap(), (Address::with_last_byte(1), entry_1));
814
815        // Check the tx view - it correctly holds entry_1
816        assert_eq!(
817            tx.cursor_dup_read::<PlainStorageState>()
818                .unwrap()
819                .walk(None)
820                .unwrap()
821                .collect::<Result<Vec<_>, _>>()
822                .unwrap(),
823            vec![
824                (Address::with_last_byte(1), entry_1), // This is ok - we removed entry_0
825            ]
826        );
827
828        // Check the remainder of walker
829        assert!(walker.next().is_none());
830    }
831
832    #[test]
833    fn db_cursor_walk() {
834        let (_tempdir, env) = create_test_db(DatabaseEnvKind::RW);
835
836        let value = Header::default();
837        let key = 1u64;
838
839        // PUT
840        let tx = env.tx_mut().expect(ERROR_INIT_TX);
841        tx.put::<Headers>(key, value.clone()).expect(ERROR_PUT);
842        tx.commit().expect(ERROR_COMMIT);
843
844        // Cursor
845        let tx = env.tx().expect(ERROR_INIT_TX);
846        let mut cursor = tx.cursor_read::<Headers>().unwrap();
847
848        let first = cursor.first().unwrap();
849        assert!(first.is_some(), "First should be our put");
850
851        // Walk
852        let walk = cursor.walk(Some(key)).unwrap();
853        let first = walk.into_iter().next().unwrap().unwrap();
854        assert_eq!(first.1, value, "First next should be put value");
855    }
856
857    #[test]
858    fn db_cursor_walk_range() {
859        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
860
861        // PUT (0, 0), (1, 0), (2, 0), (3, 0)
862        let tx = db.tx_mut().expect(ERROR_INIT_TX);
863        vec![0, 1, 2, 3]
864            .into_iter()
865            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
866            .expect(ERROR_PUT);
867        tx.commit().expect(ERROR_COMMIT);
868
869        let tx = db.tx().expect(ERROR_INIT_TX);
870        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
871
872        // [1, 3)
873        let mut walker = cursor.walk_range(1..3).unwrap();
874        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
875        assert_eq!(walker.next().unwrap().unwrap(), (2, B256::ZERO));
876        assert!(walker.next().is_none());
877        // next() returns None after walker is done
878        assert!(walker.next().is_none());
879
880        // [1, 2]
881        let mut walker = cursor.walk_range(1..=2).unwrap();
882        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
883        assert_eq!(walker.next().unwrap().unwrap(), (2, B256::ZERO));
884        // next() returns None after walker is done
885        assert!(walker.next().is_none());
886
887        // [1, ∞)
888        let mut walker = cursor.walk_range(1..).unwrap();
889        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
890        assert_eq!(walker.next().unwrap().unwrap(), (2, B256::ZERO));
891        assert_eq!(walker.next().unwrap().unwrap(), (3, B256::ZERO));
892        // next() returns None after walker is done
893        assert!(walker.next().is_none());
894
895        // [2, 4)
896        let mut walker = cursor.walk_range(2..4).unwrap();
897        assert_eq!(walker.next().unwrap().unwrap(), (2, B256::ZERO));
898        assert_eq!(walker.next().unwrap().unwrap(), (3, B256::ZERO));
899        assert!(walker.next().is_none());
900        // next() returns None after walker is done
901        assert!(walker.next().is_none());
902
903        // (∞, 3)
904        let mut walker = cursor.walk_range(..3).unwrap();
905        assert_eq!(walker.next().unwrap().unwrap(), (0, B256::ZERO));
906        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
907        assert_eq!(walker.next().unwrap().unwrap(), (2, B256::ZERO));
908        // next() returns None after walker is done
909        assert!(walker.next().is_none());
910
911        // (∞, ∞)
912        let mut walker = cursor.walk_range(..).unwrap();
913        assert_eq!(walker.next().unwrap().unwrap(), (0, B256::ZERO));
914        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
915        assert_eq!(walker.next().unwrap().unwrap(), (2, B256::ZERO));
916        assert_eq!(walker.next().unwrap().unwrap(), (3, B256::ZERO));
917        // next() returns None after walker is done
918        assert!(walker.next().is_none());
919    }
920
921    #[test]
922    fn db_cursor_walk_range_on_dup_table() {
923        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
924
925        let address0 = Address::ZERO;
926        let address1 = Address::with_last_byte(1);
927        let address2 = Address::with_last_byte(2);
928
929        let tx = db.tx_mut().expect(ERROR_INIT_TX);
930        tx.put::<AccountChangeSets>(0, AccountBeforeTx { address: address0, info: None })
931            .expect(ERROR_PUT);
932        tx.put::<AccountChangeSets>(0, AccountBeforeTx { address: address1, info: None })
933            .expect(ERROR_PUT);
934        tx.put::<AccountChangeSets>(0, AccountBeforeTx { address: address2, info: None })
935            .expect(ERROR_PUT);
936        tx.put::<AccountChangeSets>(1, AccountBeforeTx { address: address0, info: None })
937            .expect(ERROR_PUT);
938        tx.put::<AccountChangeSets>(1, AccountBeforeTx { address: address1, info: None })
939            .expect(ERROR_PUT);
940        tx.put::<AccountChangeSets>(1, AccountBeforeTx { address: address2, info: None })
941            .expect(ERROR_PUT);
942        tx.put::<AccountChangeSets>(2, AccountBeforeTx { address: address0, info: None }) // <- should not be returned by the walker
943            .expect(ERROR_PUT);
944        tx.commit().expect(ERROR_COMMIT);
945
946        let tx = db.tx().expect(ERROR_INIT_TX);
947        let mut cursor = tx.cursor_read::<AccountChangeSets>().unwrap();
948
949        let entries = cursor.walk_range(..).unwrap().collect::<Result<Vec<_>, _>>().unwrap();
950        assert_eq!(entries.len(), 7);
951
952        let mut walker = cursor.walk_range(0..=1).unwrap();
953        assert_eq!(
954            walker.next().unwrap().unwrap(),
955            (0, AccountBeforeTx { address: address0, info: None })
956        );
957        assert_eq!(
958            walker.next().unwrap().unwrap(),
959            (0, AccountBeforeTx { address: address1, info: None })
960        );
961        assert_eq!(
962            walker.next().unwrap().unwrap(),
963            (0, AccountBeforeTx { address: address2, info: None })
964        );
965        assert_eq!(
966            walker.next().unwrap().unwrap(),
967            (1, AccountBeforeTx { address: address0, info: None })
968        );
969        assert_eq!(
970            walker.next().unwrap().unwrap(),
971            (1, AccountBeforeTx { address: address1, info: None })
972        );
973        assert_eq!(
974            walker.next().unwrap().unwrap(),
975            (1, AccountBeforeTx { address: address2, info: None })
976        );
977        assert!(walker.next().is_none());
978    }
979
980    #[expect(clippy::reversed_empty_ranges)]
981    #[test]
982    fn db_cursor_walk_range_invalid() {
983        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
984
985        // PUT (0, 0), (1, 0), (2, 0), (3, 0)
986        let tx = db.tx_mut().expect(ERROR_INIT_TX);
987        vec![0, 1, 2, 3]
988            .into_iter()
989            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
990            .expect(ERROR_PUT);
991        tx.commit().expect(ERROR_COMMIT);
992
993        let tx = db.tx().expect(ERROR_INIT_TX);
994        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
995
996        // start bound greater than end bound
997        let mut res = cursor.walk_range(3..1).unwrap();
998        assert!(res.next().is_none());
999
1000        // start bound greater than end bound
1001        let mut res = cursor.walk_range(15..=2).unwrap();
1002        assert!(res.next().is_none());
1003
1004        // returning nothing
1005        let mut walker = cursor.walk_range(1..1).unwrap();
1006        assert!(walker.next().is_none());
1007    }
1008
1009    #[test]
1010    fn db_walker() {
1011        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1012
1013        // PUT (0, 0), (1, 0), (3, 0)
1014        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1015        vec![0, 1, 3]
1016            .into_iter()
1017            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1018            .expect(ERROR_PUT);
1019        tx.commit().expect(ERROR_COMMIT);
1020
1021        let tx = db.tx().expect(ERROR_INIT_TX);
1022        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1023
1024        let mut walker = Walker::new(&mut cursor, None);
1025
1026        assert_eq!(walker.next().unwrap().unwrap(), (0, B256::ZERO));
1027        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
1028        assert_eq!(walker.next().unwrap().unwrap(), (3, B256::ZERO));
1029        assert!(walker.next().is_none());
1030
1031        // transform to ReverseWalker
1032        let mut reverse_walker = walker.rev();
1033        assert_eq!(reverse_walker.next().unwrap().unwrap(), (3, B256::ZERO));
1034        assert_eq!(reverse_walker.next().unwrap().unwrap(), (1, B256::ZERO));
1035        assert_eq!(reverse_walker.next().unwrap().unwrap(), (0, B256::ZERO));
1036        assert!(reverse_walker.next().is_none());
1037    }
1038
1039    #[test]
1040    fn db_reverse_walker() {
1041        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1042
1043        // PUT (0, 0), (1, 0), (3, 0)
1044        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1045        vec![0, 1, 3]
1046            .into_iter()
1047            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1048            .expect(ERROR_PUT);
1049        tx.commit().expect(ERROR_COMMIT);
1050
1051        let tx = db.tx().expect(ERROR_INIT_TX);
1052        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1053
1054        let mut reverse_walker = ReverseWalker::new(&mut cursor, None);
1055
1056        assert_eq!(reverse_walker.next().unwrap().unwrap(), (3, B256::ZERO));
1057        assert_eq!(reverse_walker.next().unwrap().unwrap(), (1, B256::ZERO));
1058        assert_eq!(reverse_walker.next().unwrap().unwrap(), (0, B256::ZERO));
1059        assert!(reverse_walker.next().is_none());
1060
1061        // transform to Walker
1062        let mut walker = reverse_walker.forward();
1063        assert_eq!(walker.next().unwrap().unwrap(), (0, B256::ZERO));
1064        assert_eq!(walker.next().unwrap().unwrap(), (1, B256::ZERO));
1065        assert_eq!(walker.next().unwrap().unwrap(), (3, B256::ZERO));
1066        assert!(walker.next().is_none());
1067    }
1068
1069    #[test]
1070    fn db_walk_back() {
1071        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1072
1073        // PUT (0, 0), (1, 0), (3, 0)
1074        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1075        vec![0, 1, 3]
1076            .into_iter()
1077            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1078            .expect(ERROR_PUT);
1079        tx.commit().expect(ERROR_COMMIT);
1080
1081        let tx = db.tx().expect(ERROR_INIT_TX);
1082        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1083
1084        let mut reverse_walker = cursor.walk_back(Some(1)).unwrap();
1085        assert_eq!(reverse_walker.next().unwrap().unwrap(), (1, B256::ZERO));
1086        assert_eq!(reverse_walker.next().unwrap().unwrap(), (0, B256::ZERO));
1087        assert!(reverse_walker.next().is_none());
1088
1089        let mut reverse_walker = cursor.walk_back(Some(2)).unwrap();
1090        assert_eq!(reverse_walker.next().unwrap().unwrap(), (3, B256::ZERO));
1091        assert_eq!(reverse_walker.next().unwrap().unwrap(), (1, B256::ZERO));
1092        assert_eq!(reverse_walker.next().unwrap().unwrap(), (0, B256::ZERO));
1093        assert!(reverse_walker.next().is_none());
1094
1095        let mut reverse_walker = cursor.walk_back(Some(4)).unwrap();
1096        assert_eq!(reverse_walker.next().unwrap().unwrap(), (3, B256::ZERO));
1097        assert_eq!(reverse_walker.next().unwrap().unwrap(), (1, B256::ZERO));
1098        assert_eq!(reverse_walker.next().unwrap().unwrap(), (0, B256::ZERO));
1099        assert!(reverse_walker.next().is_none());
1100
1101        let mut reverse_walker = cursor.walk_back(None).unwrap();
1102        assert_eq!(reverse_walker.next().unwrap().unwrap(), (3, B256::ZERO));
1103        assert_eq!(reverse_walker.next().unwrap().unwrap(), (1, B256::ZERO));
1104        assert_eq!(reverse_walker.next().unwrap().unwrap(), (0, B256::ZERO));
1105        assert!(reverse_walker.next().is_none());
1106    }
1107
1108    #[test]
1109    fn db_cursor_seek_exact_or_previous_key() {
1110        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1111
1112        // PUT
1113        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1114        vec![0, 1, 3]
1115            .into_iter()
1116            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1117            .expect(ERROR_PUT);
1118        tx.commit().expect(ERROR_COMMIT);
1119
1120        // Cursor
1121        let missing_key = 2;
1122        let tx = db.tx().expect(ERROR_INIT_TX);
1123        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1124        assert!(cursor.current().unwrap().is_none());
1125
1126        // Seek exact
1127        let exact = cursor.seek_exact(missing_key).unwrap();
1128        assert_eq!(exact, None);
1129        assert!(cursor.current().unwrap().is_none());
1130    }
1131
1132    #[test]
1133    fn db_cursor_insert() {
1134        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1135
1136        // PUT
1137        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1138        vec![0, 1, 3, 4, 5]
1139            .into_iter()
1140            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1141            .expect(ERROR_PUT);
1142        tx.commit().expect(ERROR_COMMIT);
1143
1144        let key_to_insert = 2;
1145        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1146        let mut cursor = tx.cursor_write::<CanonicalHeaders>().unwrap();
1147
1148        // INSERT
1149        assert!(cursor.insert(key_to_insert, &B256::ZERO).is_ok());
1150        assert_eq!(cursor.current().unwrap(), Some((key_to_insert, B256::ZERO)));
1151        // INSERT (failure)
1152        assert!(matches!(
1153        cursor.insert(key_to_insert, &B256::ZERO).unwrap_err(),
1154        DatabaseError::Write(err) if *err == DatabaseWriteError {
1155            info: Error::KeyExist.into(),
1156            operation: DatabaseWriteOperation::CursorInsert,
1157            table_name: CanonicalHeaders::NAME,
1158            key: key_to_insert.encode().into(),
1159        }));
1160        assert_eq!(cursor.current().unwrap(), Some((key_to_insert, B256::ZERO)));
1161
1162        tx.commit().expect(ERROR_COMMIT);
1163
1164        // Confirm the result
1165        let tx = db.tx().expect(ERROR_INIT_TX);
1166        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1167        let res = cursor.walk(None).unwrap().map(|res| res.unwrap().0).collect::<Vec<_>>();
1168        assert_eq!(res, vec![0, 1, 2, 3, 4, 5]);
1169        tx.commit().expect(ERROR_COMMIT);
1170    }
1171
1172    #[test]
1173    fn db_cursor_insert_dup() {
1174        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1175        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1176
1177        let mut dup_cursor = tx.cursor_dup_write::<PlainStorageState>().unwrap();
1178        let key = Address::random();
1179        let subkey1 = B256::random();
1180        let subkey2 = B256::random();
1181
1182        let entry1 = StorageEntry { key: subkey1, value: U256::ZERO };
1183        assert!(dup_cursor.insert(key, &entry1).is_ok());
1184
1185        // Can't insert
1186        let entry2 = StorageEntry { key: subkey2, value: U256::ZERO };
1187        assert!(dup_cursor.insert(key, &entry2).is_err());
1188    }
1189
1190    #[test]
1191    fn db_cursor_delete_current_non_existent() {
1192        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1193        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1194
1195        let key1 = Address::with_last_byte(1);
1196        let key2 = Address::with_last_byte(2);
1197        let key3 = Address::with_last_byte(3);
1198        let mut cursor = tx.cursor_write::<PlainAccountState>().unwrap();
1199
1200        assert!(cursor.insert(key1, &Account::default()).is_ok());
1201        assert!(cursor.insert(key2, &Account::default()).is_ok());
1202        assert!(cursor.insert(key3, &Account::default()).is_ok());
1203
1204        // Seek & delete key2
1205        cursor.seek_exact(key2).unwrap();
1206        assert!(cursor.delete_current().is_ok());
1207        assert!(cursor.seek_exact(key2).unwrap().is_none());
1208
1209        // Seek & delete key2 again
1210        assert!(cursor.seek_exact(key2).unwrap().is_none());
1211        assert!(matches!(
1212            cursor.delete_current().unwrap_err(),
1213            DatabaseError::Delete(err) if err == reth_libmdbx::Error::NoData.into()));
1214        // Assert that key1 is still there
1215        assert_eq!(cursor.seek_exact(key1).unwrap(), Some((key1, Account::default())));
1216        // Assert that key3 is still there
1217        assert_eq!(cursor.seek_exact(key3).unwrap(), Some((key3, Account::default())));
1218    }
1219
1220    #[test]
1221    fn db_cursor_insert_wherever_cursor_is() {
1222        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1223        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1224
1225        // PUT
1226        vec![0, 1, 3, 5, 7, 9]
1227            .into_iter()
1228            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1229            .expect(ERROR_PUT);
1230        tx.commit().expect(ERROR_COMMIT);
1231
1232        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1233        let mut cursor = tx.cursor_write::<CanonicalHeaders>().unwrap();
1234
1235        // INSERT (cursor starts at last)
1236        cursor.last().unwrap();
1237        assert_eq!(cursor.current().unwrap(), Some((9, B256::ZERO)));
1238
1239        for pos in (2..=8).step_by(2) {
1240            assert!(cursor.insert(pos, &B256::ZERO).is_ok());
1241            assert_eq!(cursor.current().unwrap(), Some((pos, B256::ZERO)));
1242        }
1243        tx.commit().expect(ERROR_COMMIT);
1244
1245        // Confirm the result
1246        let tx = db.tx().expect(ERROR_INIT_TX);
1247        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1248        let res = cursor.walk(None).unwrap().map(|res| res.unwrap().0).collect::<Vec<_>>();
1249        assert_eq!(res, vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
1250        tx.commit().expect(ERROR_COMMIT);
1251    }
1252
1253    #[test]
1254    fn db_cursor_append() {
1255        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1256
1257        // PUT
1258        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1259        vec![0, 1, 2, 3, 4]
1260            .into_iter()
1261            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1262            .expect(ERROR_PUT);
1263        tx.commit().expect(ERROR_COMMIT);
1264
1265        // APPEND
1266        let key_to_append = 5;
1267        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1268        let mut cursor = tx.cursor_write::<CanonicalHeaders>().unwrap();
1269        assert!(cursor.append(key_to_append, &B256::ZERO).is_ok());
1270        tx.commit().expect(ERROR_COMMIT);
1271
1272        // Confirm the result
1273        let tx = db.tx().expect(ERROR_INIT_TX);
1274        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1275        let res = cursor.walk(None).unwrap().map(|res| res.unwrap().0).collect::<Vec<_>>();
1276        assert_eq!(res, vec![0, 1, 2, 3, 4, 5]);
1277        tx.commit().expect(ERROR_COMMIT);
1278    }
1279
1280    #[test]
1281    fn db_cursor_append_failure() {
1282        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1283
1284        // PUT
1285        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1286        vec![0, 1, 3, 4, 5]
1287            .into_iter()
1288            .try_for_each(|key| tx.put::<CanonicalHeaders>(key, B256::ZERO))
1289            .expect(ERROR_PUT);
1290        tx.commit().expect(ERROR_COMMIT);
1291
1292        // APPEND
1293        let key_to_append = 2;
1294        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1295        let mut cursor = tx.cursor_write::<CanonicalHeaders>().unwrap();
1296        assert!(matches!(
1297        cursor.append(key_to_append, &B256::ZERO).unwrap_err(),
1298        DatabaseError::Write(err) if *err == DatabaseWriteError {
1299            info: Error::KeyMismatch.into(),
1300            operation: DatabaseWriteOperation::CursorAppend,
1301            table_name: CanonicalHeaders::NAME,
1302            key: key_to_append.encode().into(),
1303        }));
1304        assert_eq!(cursor.current().unwrap(), Some((5, B256::ZERO))); // the end of table
1305        tx.commit().expect(ERROR_COMMIT);
1306
1307        // Confirm the result
1308        let tx = db.tx().expect(ERROR_INIT_TX);
1309        let mut cursor = tx.cursor_read::<CanonicalHeaders>().unwrap();
1310        let res = cursor.walk(None).unwrap().map(|res| res.unwrap().0).collect::<Vec<_>>();
1311        assert_eq!(res, vec![0, 1, 3, 4, 5]);
1312        tx.commit().expect(ERROR_COMMIT);
1313    }
1314
1315    #[test]
1316    fn db_cursor_upsert() {
1317        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1318        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1319
1320        let mut cursor = tx.cursor_write::<PlainAccountState>().unwrap();
1321        let key = Address::random();
1322
1323        let account = Account::default();
1324        cursor.upsert(key, &account).expect(ERROR_UPSERT);
1325        assert_eq!(cursor.seek_exact(key).unwrap(), Some((key, account)));
1326
1327        let account = Account { nonce: 1, ..Default::default() };
1328        cursor.upsert(key, &account).expect(ERROR_UPSERT);
1329        assert_eq!(cursor.seek_exact(key).unwrap(), Some((key, account)));
1330
1331        let account = Account { nonce: 2, ..Default::default() };
1332        cursor.upsert(key, &account).expect(ERROR_UPSERT);
1333        assert_eq!(cursor.seek_exact(key).unwrap(), Some((key, account)));
1334
1335        let mut dup_cursor = tx.cursor_dup_write::<PlainStorageState>().unwrap();
1336        let subkey = B256::random();
1337
1338        let value = U256::ONE;
1339        let entry1 = StorageEntry { key: subkey, value };
1340        dup_cursor.upsert(key, &entry1).expect(ERROR_UPSERT);
1341        assert_eq!(dup_cursor.seek_by_key_subkey(key, subkey).unwrap(), Some(entry1));
1342
1343        let value = U256::from(2);
1344        let entry2 = StorageEntry { key: subkey, value };
1345        dup_cursor.upsert(key, &entry2).expect(ERROR_UPSERT);
1346        assert_eq!(dup_cursor.seek_by_key_subkey(key, subkey).unwrap(), Some(entry1));
1347        assert_eq!(dup_cursor.next_dup_val().unwrap(), Some(entry2));
1348    }
1349
1350    #[test]
1351    fn db_cursor_dupsort_append() {
1352        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1353
1354        let transition_id = 2;
1355
1356        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1357        let mut cursor = tx.cursor_write::<AccountChangeSets>().unwrap();
1358        vec![0, 1, 3, 4, 5]
1359            .into_iter()
1360            .try_for_each(|val| {
1361                cursor.append(
1362                    transition_id,
1363                    &AccountBeforeTx { address: Address::with_last_byte(val), info: None },
1364                )
1365            })
1366            .expect(ERROR_APPEND);
1367        tx.commit().expect(ERROR_COMMIT);
1368
1369        // APPEND DUP & APPEND
1370        let subkey_to_append = 2;
1371        let tx = db.tx_mut().expect(ERROR_INIT_TX);
1372        let mut cursor = tx.cursor_write::<AccountChangeSets>().unwrap();
1373        assert!(matches!(
1374        cursor
1375            .append_dup(
1376                transition_id,
1377                AccountBeforeTx {
1378                    address: Address::with_last_byte(subkey_to_append),
1379                    info: None
1380                }
1381            )
1382            .unwrap_err(),
1383        DatabaseError::Write(err) if *err == DatabaseWriteError {
1384            info: Error::KeyMismatch.into(),
1385            operation: DatabaseWriteOperation::CursorAppendDup,
1386            table_name: AccountChangeSets::NAME,
1387            key: transition_id.encode().into(),
1388        }));
1389        assert!(matches!(
1390            cursor
1391                .append(
1392                    transition_id - 1,
1393                    &AccountBeforeTx {
1394                        address: Address::with_last_byte(subkey_to_append),
1395                        info: None
1396                    }
1397                )
1398                .unwrap_err(),
1399            DatabaseError::Write(err) if *err == DatabaseWriteError {
1400                info: Error::KeyMismatch.into(),
1401                operation: DatabaseWriteOperation::CursorAppend,
1402                table_name: AccountChangeSets::NAME,
1403                key: (transition_id - 1).encode().into(),
1404            }
1405        ));
1406        assert!(cursor
1407            .append(
1408                transition_id,
1409                &AccountBeforeTx { address: Address::with_last_byte(subkey_to_append), info: None }
1410            )
1411            .is_ok());
1412    }
1413
1414    #[test]
1415    #[allow(clippy::needless_update)]
1416    #[allow(clippy::clone_on_copy)]
1417    fn db_closure_put_get() {
1418        let tempdir = TempDir::new().expect(ERROR_TEMPDIR);
1419        let path = tempdir.path();
1420
1421        let value = Account {
1422            nonce: 18446744073709551615,
1423            bytecode_hash: Some(B256::random()),
1424            balance: U256::MAX,
1425            ..Default::default()
1426        };
1427        let key = address!("0xa2c122be93b0074270ebee7f6b7292c7deb45047");
1428
1429        {
1430            let env = create_test_db_with_path(DatabaseEnvKind::RW, path);
1431
1432            // PUT
1433            let result = env.update(|tx| {
1434                tx.put::<PlainAccountState>(key, value.clone()).expect(ERROR_PUT);
1435                200
1436            });
1437            assert_eq!(result.expect(ERROR_RETURN_VALUE), 200);
1438        }
1439
1440        let env = DatabaseEnv::open(
1441            path,
1442            DatabaseEnvKind::RO,
1443            DatabaseArguments::new(ClientVersion::default()),
1444        )
1445        .expect(ERROR_DB_CREATION);
1446
1447        // GET
1448        let result =
1449            env.view(|tx| tx.get::<PlainAccountState>(key).expect(ERROR_GET)).expect(ERROR_GET);
1450
1451        assert_eq!(result, Some(value))
1452    }
1453
1454    #[test]
1455    fn db_dup_sort() {
1456        let (_tempdir, env) = create_test_db(DatabaseEnvKind::RW);
1457        let key = address!("0xa2c122be93b0074270ebee7f6b7292c7deb45047");
1458
1459        // PUT (0,0)
1460        let value00 = StorageEntry::default();
1461        env.update(|tx| tx.put::<PlainStorageState>(key, value00).expect(ERROR_PUT)).unwrap();
1462
1463        // PUT (2,2)
1464        let value22 = StorageEntry { key: B256::with_last_byte(2), value: U256::from(2) };
1465        env.update(|tx| tx.put::<PlainStorageState>(key, value22).expect(ERROR_PUT)).unwrap();
1466
1467        // PUT (1,1)
1468        let value11 = StorageEntry { key: B256::with_last_byte(1), value: U256::ONE };
1469        env.update(|tx| tx.put::<PlainStorageState>(key, value11).expect(ERROR_PUT)).unwrap();
1470
1471        // Iterate with cursor
1472        {
1473            let tx = env.tx().expect(ERROR_INIT_TX);
1474            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1475
1476            // Notice that value11 and value22 have been ordered in the DB.
1477            assert_eq!(Some(value00), cursor.next_dup_val().unwrap());
1478            assert_eq!(Some(value11), cursor.next_dup_val().unwrap());
1479            assert_eq!(Some(value22), cursor.next_dup_val().unwrap());
1480        }
1481
1482        // Seek value with exact subkey
1483        {
1484            let tx = env.tx().expect(ERROR_INIT_TX);
1485            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1486            let mut walker = cursor.walk_dup(Some(key), Some(B256::with_last_byte(1))).unwrap();
1487            assert_eq!(
1488                (key, value11),
1489                walker
1490                    .next()
1491                    .expect("element should exist.")
1492                    .expect("should be able to retrieve it.")
1493            );
1494        }
1495    }
1496
1497    #[test]
1498    fn db_walk_dup_with_not_existing_key() {
1499        let (_tempdir, env) = create_test_db(DatabaseEnvKind::RW);
1500        let key = address!("0xa2c122be93b0074270ebee7f6b7292c7deb45047");
1501
1502        // PUT (0,0)
1503        let value00 = StorageEntry::default();
1504        env.update(|tx| tx.put::<PlainStorageState>(key, value00).expect(ERROR_PUT)).unwrap();
1505
1506        // PUT (2,2)
1507        let value22 = StorageEntry { key: B256::with_last_byte(2), value: U256::from(2) };
1508        env.update(|tx| tx.put::<PlainStorageState>(key, value22).expect(ERROR_PUT)).unwrap();
1509
1510        // PUT (1,1)
1511        let value11 = StorageEntry { key: B256::with_last_byte(1), value: U256::ONE };
1512        env.update(|tx| tx.put::<PlainStorageState>(key, value11).expect(ERROR_PUT)).unwrap();
1513
1514        // Try to walk_dup with not existing key should immediately return None
1515        {
1516            let tx = env.tx().expect(ERROR_INIT_TX);
1517            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1518            let not_existing_key = Address::ZERO;
1519            let mut walker = cursor.walk_dup(Some(not_existing_key), None).unwrap();
1520            assert!(walker.next().is_none());
1521        }
1522    }
1523
1524    #[test]
1525    fn db_iterate_over_all_dup_values() {
1526        let (_tempdir, env) = create_test_db(DatabaseEnvKind::RW);
1527        let key1 = Address::repeat_byte(0x11);
1528        let key2 = Address::repeat_byte(0x22);
1529
1530        // PUT key1 (0,0)
1531        let value00 = StorageEntry::default();
1532        env.update(|tx| tx.put::<PlainStorageState>(key1, value00).expect(ERROR_PUT)).unwrap();
1533
1534        // PUT key1 (1,1)
1535        let value11 = StorageEntry { key: B256::with_last_byte(1), value: U256::ONE };
1536        env.update(|tx| tx.put::<PlainStorageState>(key1, value11).expect(ERROR_PUT)).unwrap();
1537
1538        // PUT key2 (2,2)
1539        let value22 = StorageEntry { key: B256::with_last_byte(2), value: U256::from(2) };
1540        env.update(|tx| tx.put::<PlainStorageState>(key2, value22).expect(ERROR_PUT)).unwrap();
1541
1542        // Iterate with walk_dup
1543        {
1544            let tx = env.tx().expect(ERROR_INIT_TX);
1545            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1546            let mut walker = cursor.walk_dup(None, None).unwrap();
1547
1548            // Notice that value11 and value22 have been ordered in the DB.
1549            assert_eq!((key1, value00), walker.next().unwrap().unwrap());
1550            assert_eq!((key1, value11), walker.next().unwrap().unwrap());
1551            // NOTE: Dup cursor does NOT iterates on all values but only on duplicated values of the
1552            // same key. assert_eq!(Ok(Some(value22.clone())), walker.next());
1553            assert!(walker.next().is_none());
1554        }
1555
1556        // Iterate by using `walk`
1557        {
1558            let tx = env.tx().expect(ERROR_INIT_TX);
1559            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1560            let first = cursor.first().unwrap().unwrap();
1561            let mut walker = cursor.walk(Some(first.0)).unwrap();
1562            assert_eq!((key1, value00), walker.next().unwrap().unwrap());
1563            assert_eq!((key1, value11), walker.next().unwrap().unwrap());
1564            assert_eq!((key2, value22), walker.next().unwrap().unwrap());
1565        }
1566    }
1567
1568    #[test]
1569    fn dup_value_with_same_subkey() {
1570        let (_tempdir, env) = create_test_db(DatabaseEnvKind::RW);
1571        let key1 = Address::repeat_byte(0x11);
1572        let key2 = Address::repeat_byte(0x22);
1573
1574        // PUT key1 (0,1)
1575        let value01 = StorageEntry { key: B256::with_last_byte(0), value: U256::ONE };
1576        env.update(|tx| tx.put::<PlainStorageState>(key1, value01).expect(ERROR_PUT)).unwrap();
1577
1578        // PUT key1 (0,0)
1579        let value00 = StorageEntry::default();
1580        env.update(|tx| tx.put::<PlainStorageState>(key1, value00).expect(ERROR_PUT)).unwrap();
1581
1582        // PUT key2 (2,2)
1583        let value22 = StorageEntry { key: B256::with_last_byte(2), value: U256::from(2) };
1584        env.update(|tx| tx.put::<PlainStorageState>(key2, value22).expect(ERROR_PUT)).unwrap();
1585
1586        // Iterate with walk
1587        {
1588            let tx = env.tx().expect(ERROR_INIT_TX);
1589            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1590            let first = cursor.first().unwrap().unwrap();
1591            let mut walker = cursor.walk(Some(first.0)).unwrap();
1592
1593            // NOTE: Both values are present
1594            assert_eq!((key1, value00), walker.next().unwrap().unwrap());
1595            assert_eq!((key1, value01), walker.next().unwrap().unwrap());
1596            assert_eq!((key2, value22), walker.next().unwrap().unwrap());
1597        }
1598
1599        // seek_by_key_subkey
1600        {
1601            let tx = env.tx().expect(ERROR_INIT_TX);
1602            let mut cursor = tx.cursor_dup_read::<PlainStorageState>().unwrap();
1603
1604            // NOTE: There are two values with same SubKey but only first one is shown
1605            assert_eq!(value00, cursor.seek_by_key_subkey(key1, value00.key).unwrap().unwrap());
1606            // key1 but value is greater than the one in the DB
1607            assert_eq!(None, cursor.seek_by_key_subkey(key1, value22.key).unwrap());
1608        }
1609    }
1610
1611    #[test]
1612    fn db_sharded_key() {
1613        let (_tempdir, db) = create_test_db(DatabaseEnvKind::RW);
1614        let real_key = address!("0xa2c122be93b0074270ebee7f6b7292c7deb45047");
1615
1616        let shards = 5;
1617        for i in 1..=shards {
1618            let key = ShardedKey::new(real_key, if i == shards { u64::MAX } else { i * 100 });
1619            let list = IntegerList::new_pre_sorted([i * 100u64]);
1620
1621            db.update(|tx| tx.put::<AccountsHistory>(key.clone(), list.clone()).expect(""))
1622                .unwrap();
1623        }
1624
1625        // Seek value with non existing key.
1626        {
1627            let tx = db.tx().expect(ERROR_INIT_TX);
1628            let mut cursor = tx.cursor_read::<AccountsHistory>().unwrap();
1629
1630            // It will seek the one greater or equal to the query. Since we have `Address | 100`,
1631            // `Address | 200` in the database and we're querying `Address | 150` it will return us
1632            // `Address | 200`.
1633            let mut walker = cursor.walk(Some(ShardedKey::new(real_key, 150))).unwrap();
1634            let (key, list) = walker
1635                .next()
1636                .expect("element should exist.")
1637                .expect("should be able to retrieve it.");
1638
1639            assert_eq!(ShardedKey::new(real_key, 200), key);
1640            let list200 = IntegerList::new_pre_sorted([200u64]);
1641            assert_eq!(list200, list);
1642        }
1643        // Seek greatest index
1644        {
1645            let tx = db.tx().expect(ERROR_INIT_TX);
1646            let mut cursor = tx.cursor_read::<AccountsHistory>().unwrap();
1647
1648            // It will seek the MAX value of transition index and try to use prev to get first
1649            // biggers.
1650            let _unknown = cursor.seek_exact(ShardedKey::new(real_key, u64::MAX)).unwrap();
1651            let (key, list) = cursor
1652                .prev()
1653                .expect("element should exist.")
1654                .expect("should be able to retrieve it.");
1655
1656            assert_eq!(ShardedKey::new(real_key, 400), key);
1657            let list400 = IntegerList::new_pre_sorted([400u64]);
1658            assert_eq!(list400, list);
1659        }
1660    }
1661}