feat(storage): use mdbx_txn_reset to time out transactions

crates/storage/db/src/implementation/mdbx/mod.rs

@@ -123,17 +123,19 @@ impl Database for DatabaseEnv {
     type TXMut = tx::Tx<RW>;
 
     fn tx(&self) -> Result<Self::TX, DatabaseError> {
-        Ok(Tx::new_with_metrics(
+        Tx::new_with_metrics(
             self.inner.begin_ro_txn().map_err(|e| DatabaseError::InitTx(e.into()))?,
             self.metrics.as_ref().cloned(),
-        ))
+        )
+        .map_err(|e| DatabaseError::InitTx(e.into()))
     }
 
     fn tx_mut(&self) -> Result<Self::TXMut, DatabaseError> {
-        Ok(Tx::new_with_metrics(
+        Tx::new_with_metrics(
             self.inner.begin_rw_txn().map_err(|e| DatabaseError::InitTx(e.into()))?,
             self.metrics.as_ref().cloned(),
-        ))
+        )
+        .map_err(|e| DatabaseError::InitTx(e.into()))
     }
 }
 

crates/storage/db/src/implementation/mdbx/tx.rs

@@ -53,14 +53,16 @@ impl<K: TransactionKind> Tx<K> {
     pub fn new_with_metrics(
         inner: Transaction<K>,
         env_metrics: Option<Arc<DatabaseEnvMetrics>>,
-    ) -> Self {
-        let metrics_handler = env_metrics.map(|env_metrics| {
-            let handler = MetricsHandler::<K>::new(inner.id(), env_metrics);
-            handler.env_metrics.record_opened_transaction(handler.transaction_mode());
-            handler.log_transaction_opened();
-            handler
-        });
-        Self::new_inner(inner, metrics_handler)
+    ) -> reth_libmdbx::Result<Self> {
+        let metrics_handler = env_metrics
+            .map(|env_metrics| {
+                let handler = MetricsHandler::<K>::new(inner.id()?, env_metrics);
+                handler.env_metrics.record_opened_transaction(handler.transaction_mode());
+                handler.log_transaction_opened();
+                Ok(handler)
+            })
+            .transpose()?;
+        Ok(Self::new_inner(inner, metrics_handler))
     }
 
     #[inline]
@@ -76,8 +78,8 @@ impl<K: TransactionKind> Tx<K> {
     }
 
     /// Gets this transaction ID.
-    pub fn id(&self) -> u64 {
-        self.metrics_handler.as_ref().map_or_else(|| self.inner.id(), |handler| handler.txn_id)
+    pub fn id(&self) -> reth_libmdbx::Result<u64> {
+        self.metrics_handler.as_ref().map_or_else(|| self.inner.id(), |handler| Ok(handler.txn_id))
     }
 
     /// Gets a table database handle if it exists, otherwise creates it.

crates/storage/libmdbx-rs/benches/cursor.rs

@@ -78,27 +78,29 @@ fn bench_get_seq_raw(c: &mut Criterion) {
     let (_dir, env) = setup_bench_db(n);
 
     let dbi = env.begin_ro_txn().unwrap().open_db(None).unwrap().dbi();
-    let _txn = env.begin_ro_txn().unwrap();
-    let txn = _txn.txn();
+    let txn = env.begin_ro_txn().unwrap();
 
     let mut key = MDBX_val { iov_len: 0, iov_base: ptr::null_mut() };
     let mut data = MDBX_val { iov_len: 0, iov_base: ptr::null_mut() };
     let mut cursor: *mut MDBX_cursor = ptr::null_mut();
 
     c.bench_function("bench_get_seq_raw", |b| {
         b.iter(|| unsafe {
-            mdbx_cursor_open(txn, dbi, &mut cursor);
-            let mut i = 0;
-            let mut count = 0u32;
+            txn.txn_execute(|txn| {
+                mdbx_cursor_open(txn, dbi, &mut cursor);
+                let mut i = 0;
+                let mut count = 0u32;
 
-            while mdbx_cursor_get(cursor, &mut key, &mut data, MDBX_NEXT) == 0 {
-                i += key.iov_len + data.iov_len;
-                count += 1;
-            }
+                while mdbx_cursor_get(cursor, &mut key, &mut data, MDBX_NEXT) == 0 {
+                    i += key.iov_len + data.iov_len;
+                    count += 1;
+                }
 
-            black_box(i);
-            assert_eq!(count, n);
-            mdbx_cursor_close(cursor);
+                black_box(i);
+                assert_eq!(count, n);
+                mdbx_cursor_close(cursor);
+            })
+            .unwrap();
         })
     });
 }

crates/storage/libmdbx-rs/benches/transaction.rs

@@ -46,7 +46,7 @@ fn bench_get_rand_raw(c: &mut Criterion) {
 
     c.bench_function("bench_get_rand_raw", |b| {
         b.iter(|| unsafe {
-            txn.with_raw_tx_ptr(|txn| {
+            txn.txn_execute(|txn| {
                 let mut i: size_t = 0;
                 for key in &keys {
                     key_val.iov_len = key.len() as size_t;
@@ -57,7 +57,8 @@ fn bench_get_rand_raw(c: &mut Criterion) {
                     i += key_val.iov_len;
                 }
                 black_box(i);
-            });
+            })
+            .unwrap();
         })
     });
 }

crates/storage/libmdbx-rs/src/cursor.rs

@@ -1,5 +1,5 @@
 use crate::{
-    error::{mdbx_result, mdbx_result_with_tx_kind, Error, Result},
+    error::{mdbx_result, Error, Result},
     flags::*,
     mdbx_try_optional,
     transaction::{TransactionKind, RW},
@@ -30,26 +30,26 @@ where
     pub(crate) fn new(txn: Transaction<K>, dbi: ffi::MDBX_dbi) -> Result<Self> {
         let mut cursor: *mut ffi::MDBX_cursor = ptr::null_mut();
         unsafe {
-            mdbx_result_with_tx_kind::<K>(
-                txn.txn_execute(|txn| ffi::mdbx_cursor_open(txn, dbi, &mut cursor)),
-                txn.txn(),
-                txn.env().txn_manager(),
-            )?;
+            txn.txn_execute(|txn_ptr| {
+                mdbx_result(ffi::mdbx_cursor_open(txn_ptr, dbi, &mut cursor))
+            })??;
         }
         Ok(Self { txn, cursor })
     }
 
     fn new_at_position(other: &Self) -> Result<Self> {
         unsafe {
-            let cursor = ffi::mdbx_cursor_create(ptr::null_mut());
+            other.txn.txn_execute(|_| {
+                let cursor = ffi::mdbx_cursor_create(ptr::null_mut());
 
-            let res = ffi::mdbx_cursor_copy(other.cursor(), cursor);
+                let res = ffi::mdbx_cursor_copy(other.cursor(), cursor);
 
-            let s = Self { txn: other.txn.clone(), cursor };
+                let s = Self { txn: other.txn.clone(), cursor };
 
-            mdbx_result_with_tx_kind::<K>(res, s.txn.txn(), s.txn.env().txn_manager())?;
+                mdbx_result(res)?;
 
-            Ok(s)
+                Ok(s)
+            })?
         }
     }
 
@@ -95,11 +95,12 @@ where
             let key_ptr = key_val.iov_base;
             let data_ptr = data_val.iov_base;
             self.txn.txn_execute(|txn| {
-                let v = mdbx_result_with_tx_kind::<K>(
-                    ffi::mdbx_cursor_get(self.cursor, &mut key_val, &mut data_val, op),
-                    txn,
-                    self.txn.env().txn_manager(),
-                )?;
+                let v = mdbx_result(ffi::mdbx_cursor_get(
+                    self.cursor,
+                    &mut key_val,
+                    &mut data_val,
+                    op,
+                ))?;
                 assert_ne!(data_ptr, data_val.iov_base);
                 let key_out = {
                     // MDBX wrote in new key
@@ -111,7 +112,7 @@ where
                 };
                 let data_out = Value::decode_val::<K>(txn, data_val)?;
                 Ok((key_out, data_out, v))
-            })
+            })?
         }
     }
 
@@ -444,7 +445,7 @@ impl Cursor<RW> {
         mdbx_result(unsafe {
             self.txn.txn_execute(|_| {
                 ffi::mdbx_cursor_put(self.cursor, &key_val, &mut data_val, flags.bits())
-            })
+            })?
         })?;
 
         Ok(())
@@ -458,7 +459,7 @@ impl Cursor<RW> {
     /// current key, if the database was opened with [DatabaseFlags::DUP_SORT].
     pub fn del(&mut self, flags: WriteFlags) -> Result<()> {
         mdbx_result(unsafe {
-            self.txn.txn_execute(|_| ffi::mdbx_cursor_del(self.cursor, flags.bits()))
+            self.txn.txn_execute(|_| ffi::mdbx_cursor_del(self.cursor, flags.bits()))?
         })?;
 
         Ok(())
@@ -470,7 +471,7 @@ where
     K: TransactionKind,
 {
     fn clone(&self) -> Self {
-        self.txn.txn_execute(|_| Self::new_at_position(self).unwrap())
+        Self::new_at_position(self).unwrap()
     }
 }
 
@@ -488,7 +489,7 @@ where
     K: TransactionKind,
 {
     fn drop(&mut self) {
-        self.txn.txn_execute(|_| unsafe { ffi::mdbx_cursor_close(self.cursor) })
+        let _ = self.txn.txn_execute(|_| unsafe { ffi::mdbx_cursor_close(self.cursor) });
     }
 }
 
@@ -564,7 +565,7 @@ where
                 let mut data = ffi::MDBX_val { iov_len: 0, iov_base: ptr::null_mut() };
                 let op = mem::replace(op, *next_op);
                 unsafe {
-                    cursor.txn.txn_execute(|txn| {
+                    let result = cursor.txn.txn_execute(|txn| {
                         match ffi::mdbx_cursor_get(cursor.cursor(), &mut key, &mut data, op) {
                             ffi::MDBX_SUCCESS => {
                                 let key = match Key::decode_val::<K>(txn, key) {
@@ -583,7 +584,11 @@ where
                             ffi::MDBX_NOTFOUND | ffi::MDBX_ENODATA => None,
                             error => Some(Err(Error::from_err_code(error))),
                         }
-                    })
+                    });
+                    match result {
+                        Ok(result) => result,
+                        Err(err) => Some(Err(err)),
+                    }
                 }
             }
             Self::Err(err) => err.take().map(Err),
@@ -655,7 +660,7 @@ where
                 let mut data = ffi::MDBX_val { iov_len: 0, iov_base: ptr::null_mut() };
                 let op = mem::replace(op, *next_op);
                 unsafe {
-                    cursor.txn.txn_execute(|txn| {
+                    let result = cursor.txn.txn_execute(|txn| {
                         match ffi::mdbx_cursor_get(cursor.cursor(), &mut key, &mut data, op) {
                             ffi::MDBX_SUCCESS => {
                                 let key = match Key::decode_val::<K>(txn, key) {
@@ -674,7 +679,11 @@ where
                             ffi::MDBX_NOTFOUND | ffi::MDBX_ENODATA => None,
                             error => Some(Err(Error::from_err_code(error))),
                         }
-                    })
+                    });
+                    match result {
+                        Ok(result) => result,
+                        Err(err) => Some(Err(err)),
+                    }
                 }
             }
             Iter::Err(err) => err.take().map(Err),
@@ -752,17 +761,15 @@ where
                 let mut data = ffi::MDBX_val { iov_len: 0, iov_base: ptr::null_mut() };
                 let op = mem::replace(op, ffi::MDBX_NEXT_NODUP);
 
-                cursor.txn.txn_execute(|_| {
-                    let err_code =
-                        unsafe { ffi::mdbx_cursor_get(cursor.cursor(), &mut key, &mut data, op) };
-
-                    (err_code == ffi::MDBX_SUCCESS).then(|| {
-                        IntoIter::new(
-                            Cursor::new_at_position(&**cursor).unwrap(),
-                            ffi::MDBX_GET_CURRENT,
-                            ffi::MDBX_NEXT_DUP,
-                        )
-                    })
+                let err_code =
+                    unsafe { ffi::mdbx_cursor_get(cursor.cursor(), &mut key, &mut data, op) };
+
+                (err_code == ffi::MDBX_SUCCESS).then(|| {
+                    IntoIter::new(
+                        Cursor::new_at_position(&**cursor).unwrap(),
+                        ffi::MDBX_GET_CURRENT,
+                        ffi::MDBX_NEXT_DUP,
+                    )
                 })
             }
             IterDup::Err(err) => err.take().map(|e| IntoIter::Err(Some(e))),

crates/storage/libmdbx-rs/src/database.rs

@@ -1,5 +1,5 @@
 use crate::{
-    error::{mdbx_result_with_tx_kind, Result},
+    error::{mdbx_result, Result},
     transaction::TransactionKind,
     Environment, Transaction,
 };
@@ -31,12 +31,8 @@ impl Database {
         let name_ptr = if let Some(c_name) = &c_name { c_name.as_ptr() } else { ptr::null() };
         let mut dbi: ffi::MDBX_dbi = 0;
         txn.txn_execute(|txn_ptr| {
-            mdbx_result_with_tx_kind::<K>(
-                unsafe { ffi::mdbx_dbi_open(txn_ptr, name_ptr, flags, &mut dbi) },
-                txn_ptr,
-                txn.env().txn_manager(),
-            )
-        })?;
+            mdbx_result(unsafe { ffi::mdbx_dbi_open(txn_ptr, name_ptr, flags, &mut dbi) })
+        })??;
         Ok(Self::new_from_ptr(dbi, txn.env().clone()))
     }
 

crates/storage/libmdbx-rs/src/error.rs

@@ -1,4 +1,3 @@
-use crate::{txn_manager::TxnManager, TransactionKind};
 use libc::c_int;
 use std::result;
 
@@ -193,9 +192,10 @@ impl Error {
             Error::DecodeErrorLenDiff | Error::DecodeError => ffi::MDBX_EINVAL,
             Error::Access => ffi::MDBX_EACCESS,
             Error::TooLarge => ffi::MDBX_TOO_LARGE,
-            Error::BadSignature | Error::ReadTransactionAborted => ffi::MDBX_EBADSIGN,
+            Error::BadSignature => ffi::MDBX_EBADSIGN,
             Error::WriteTransactionUnsupportedInReadOnlyMode => ffi::MDBX_EACCESS,
             Error::NestedTransactionsUnsupportedWithWriteMap => ffi::MDBX_EACCESS,
+            Error::ReadTransactionAborted => -69000, // Custom non-MDBX error code
             Error::Other(err_code) => *err_code,
         }
     }
@@ -216,33 +216,6 @@ pub(crate) fn mdbx_result(err_code: c_int) -> Result<bool> {
     }
 }
 
-#[cfg(feature = "read-tx-timeouts")]
-#[inline]
-pub(crate) fn mdbx_result_with_tx_kind<K: TransactionKind>(
-    err_code: c_int,
-    txn: *mut ffi::MDBX_txn,
-    txn_manager: &TxnManager,
-) -> Result<bool> {
-    if K::IS_READ_ONLY &&
-        txn_manager.remove_aborted_read_transaction(txn).is_some() &&
-        err_code == ffi::MDBX_EBADSIGN
-    {
-        return Err(Error::ReadTransactionAborted)
-    }
-
-    mdbx_result(err_code)
-}
-
-#[cfg(not(feature = "read-tx-timeouts"))]
-#[inline]
-pub(crate) fn mdbx_result_with_tx_kind<K: TransactionKind>(
-    err_code: c_int,
-    _txn: *mut ffi::MDBX_txn,
-    _txn_manager: &TxnManager,
-) -> Result<bool> {
-    mdbx_result(err_code)
-}
-
 #[macro_export]
 macro_rules! mdbx_try_optional {
     ($expr:expr) => {{