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