fix(db): race condition mdbx abort tx

crates/storage/libmdbx-rs/Cargo.toml

@@ -22,7 +22,7 @@ indexmap = "2"
 libc = "0.2"
 parking_lot.workspace = true
 thiserror.workspace = true
-dashmap = { version = "5.5.3", features = ["inline"], optional = true }
+dashmap = { version = "5.5.3", features = ["inline"] }
 tracing = { workspace = true, optional = true }
 
 ffi = { package = "reth-mdbx-sys", path = "./mdbx-sys" }
@@ -33,7 +33,7 @@ libffi = "3.2.0"
 [features]
 default = []
 return-borrowed = []
-read-tx-timeouts = ["dashmap", "dashmap/inline", "tracing"]
+read-tx-timeouts = ["tracing"]
 
 [dev-dependencies]
 pprof = { workspace = true, features = ["flamegraph", "frame-pointer", "criterion"] }

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

@@ -705,16 +705,17 @@ impl EnvironmentBuilder {
 
         #[cfg(feature = "read-tx-timeouts")]
         let txn_manager = {
-            let mut txn_manager = TxnManager::new(EnvPtr(env));
+            let env_ptr = EnvPtr(env);
             if let crate::MaxReadTransactionDuration::Set(duration) = self
                 .max_read_transaction_duration
                 .unwrap_or(read_transactions::MaxReadTransactionDuration::Set(
                     DEFAULT_MAX_READ_TRANSACTION_DURATION,
                 ))
             {
-                txn_manager = txn_manager.with_max_read_transaction_duration(duration);
-            };
-            txn_manager
+                TxnManager::new_with_max_read_transaction_duration(env_ptr, duration)
+            } else {
+                TxnManager::new(env_ptr)
+            }
         };
 
         let env = EnvironmentInner { env, txn_manager, env_kind: self.kind };

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

@@ -120,6 +120,8 @@ pub enum Error {
     WriteTransactionUnsupportedInReadOnlyMode,
     #[error("read transaction has been aborted by the transaction manager")]
     ReadTransactionAborted,
+    #[error("write transaction has been aborted by the transaction manager")]
+    RWTransactionAborted,
     /// Unknown error code.
     #[error("unknown error code")]
     Other(i32),
@@ -193,7 +195,9 @@ 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 | Error::ReadTransactionAborted | Error::RWTransactionAborted => {
+                ffi::MDBX_EBADSIGN
+            }
             Error::WriteTransactionUnsupportedInReadOnlyMode => ffi::MDBX_EACCESS,
             Error::NestedTransactionsUnsupportedWithWriteMap => ffi::MDBX_EACCESS,
             Error::Other(err_code) => *err_code,

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

@@ -359,17 +359,28 @@ where
             if !self.has_committed() {
                 if K::IS_READ_ONLY {
                     #[cfg(feature = "read-tx-timeouts")]
-                    self.env.txn_manager().remove_active_read_transaction(txn);
+                    {
+                        self.env.txn_manager().remove_active_read_transaction(txn);
+
+                        let new_aborted_ro =
+                            self.env.txn_manager().add_aborted_read_transaction(txn);
+                        if let Some(false) = new_aborted_ro {
+                            return
+                        }
+                    }
 
                     unsafe {
                         ffi::mdbx_txn_abort(txn);
                     }
                 } else {
                     let (sender, rx) = sync_channel(0);
-                    self.env
-                        .txn_manager()
-                        .send_message(TxnManagerMessage::Abort { tx: TxnPtr(txn), sender });
-                    rx.recv().unwrap().unwrap();
+                    self.env.txn_manager().send_message(TxnManagerMessage::Abort {
+                        tx: TxnPtr(txn),
+                        flags: K::OPEN_FLAGS,
+                        sender,
+                    });
+                    let res = rx.recv().unwrap();
+                    assert!(res == Err(Error::RWTransactionAborted) || res.is_ok())
                 }
             }
         })