fix(db): race condition mdbx abort tx

crates/storage/libmdbx-rs/Cargo.toml

@@ -30,6 +30,10 @@ ffi = { package = "reth-mdbx-sys", path = "./mdbx-sys" }
 [target.'cfg(not(windows))'.dependencies]
 libffi = "3.2.0"
 
+# metrics
+reth-metrics.workspace = true
+metrics.workspace = true
+
 [features]
 default = []
 return-borrowed = []

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

@@ -216,6 +216,8 @@ pub(crate) fn mdbx_result(err_code: c_int) -> Result<bool> {
     }
 }
 
+/// Returns the appropriate error. Removes the transaction from aborted transactions, if present,
+/// as a side-effect.
 #[cfg(feature = "read-tx-timeouts")]
 #[inline]
 pub(crate) fn mdbx_result_with_tx_kind<K: TransactionKind>(

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

@@ -17,6 +17,7 @@ pub use crate::{
     },
     error::{Error, Result},
     flags::*,
+    metrics::*,
     transaction::{CommitLatency, Transaction, TransactionKind, RO, RW},
 };
 pub mod ffi {
@@ -32,6 +33,7 @@ mod database;
 mod environment;
 mod error;
 mod flags;
+mod metrics;
 mod transaction;
 mod txn_manager;
 

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

@@ -0,0 +1,10 @@
+use metrics::Gauge;
+use reth_metrics::Metrics;
+
+#[derive(Metrics)]
+#[metrics(scope = "storage.libmdbxrs.txn_manager")]
+/// `TxnManager` metrics.
+pub struct TxnManagerMetrics {
+    /// The number of aborted transactions that are currently tracked.
+    pub(super) aborted: Gauge,
+}

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

@@ -359,17 +359,26 @@ where
             if !self.has_committed() {
                 if K::IS_READ_ONLY {
                     #[cfg(feature = "read-tx-timeouts")]
-                    self.env.txn_manager().remove_active_read_transaction(txn);
+                    {
+                        if let Some(true) = self.env.txn_manager().is_aborted_read_transaction(txn)
+                        {
+                            return
+                        }
+
+                        self.env.txn_manager().remove_active_read_transaction(txn);
+                    }
 
                     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,
+                    });
+                    rx.recv().unwrap().unwrap().unwrap();
                 }
             }
         })

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

@@ -1,7 +1,7 @@
 use crate::{
     environment::EnvPtr,
     error::{mdbx_result, Result},
-    CommitLatency,
+    CommitLatency, Error,
 };
 use std::{
     ptr,
@@ -15,15 +15,19 @@ unsafe impl Sync for TxnPtr {}
 
 pub(crate) enum TxnManagerMessage {
     Begin { parent: TxnPtr, flags: ffi::MDBX_txn_flags_t, sender: SyncSender<Result<TxnPtr>> },
-    Abort { tx: TxnPtr, sender: SyncSender<Result<bool>> },
+    Abort { tx: TxnPtr, flags: ffi::MDBX_txn_flags_t, sender: SyncSender<Result<Result<bool>>> },
     Commit { tx: TxnPtr, sender: SyncSender<Result<(bool, CommitLatency)>> },
 }
 
 /// Manages transactions by doing two things:
-/// - Opening, aborting, and committing transactions using [TxnManager::send_message] with the
-///   corresponding [TxnManagerMessage]
-/// - Aborting long-lived read transactions (if the `read-tx-timeouts` feature is enabled and
-///   `TxnManager::with_max_read_transaction_duration` is called)
+/// - Opening, aborting, and committing transactions using [`TxnManager::send_message`] with the
+///   corresponding [TxnManagerMessage]. This is mainly since MDBX requires that write transactions
+///   are opened from the same thread.
+/// - API for tracking active and aborted long-lived read transactions (if the `read-tx-timeouts`
+///   feature is enabled and [`TxnManager::new_with_max_read_transaction_duration`] constructor is
+///   used).
+/// - Aborting long-lived read transactions in the background (if the `read-tx-timeouts` feature is
+///   enabled and [`TxnManager::new_with_max_read_transaction_duration`] constructor is used).
 #[derive(Debug)]
 pub(crate) struct TxnManager {
     sender: SyncSender<TxnManagerMessage>,
@@ -54,6 +58,8 @@ impl TxnManager {
     fn start_message_listener(&self, env: EnvPtr, rx: Receiver<TxnManagerMessage>) {
         #[cfg(feature = "read-tx-timeouts")]
         let read_transactions = self.read_transactions.clone();
+        #[cfg(feature = "read-tx-timeouts")]
+        use crate::transaction::TransactionKind;
 
         std::thread::spawn(move || {
             #[allow(clippy::redundant_locals)]
@@ -76,8 +82,6 @@ impl TxnManager {
 
                             #[cfg(feature = "read-tx-timeouts")]
                             {
-                                use crate::transaction::TransactionKind;
-
                                 if res.is_ok() && flags == crate::transaction::RO::OPEN_FLAGS {
                                     if let Some(read_transactions) = &read_transactions {
                                         read_transactions.add_active(txn);
@@ -87,13 +91,25 @@ impl TxnManager {
 
                             sender.send(res).unwrap();
                         }
-                        TxnManagerMessage::Abort { tx, sender } => {
+                        TxnManagerMessage::Abort { tx, flags, sender } => {
                             #[cfg(feature = "read-tx-timeouts")]
-                            if let Some(read_transactions) = &read_transactions {
-                                read_transactions.remove_active(tx.0);
+                            {
+                                if flags == crate::transaction::RO::OPEN_FLAGS {
+                                    if let Some(read_transactions) = &read_transactions {
+                                        if read_transactions.is_aborted(tx.0) {
+                                            sender
+                                                .send(Err(Error::ReadTransactionAborted))
+                                                .unwrap();
+                                            continue
+                                        }
+
+                                        read_transactions.remove_active(tx.0);
+                                    }
+                                }
                             }
 
-                            sender.send(mdbx_result(unsafe { ffi::mdbx_txn_abort(tx.0) })).unwrap();
+                            let res = mdbx_result(unsafe { ffi::mdbx_txn_abort(tx.0) });
+                            sender.send(Ok(res)).unwrap();
                         }
                         TxnManagerMessage::Commit { tx, sender } => {
                             #[cfg(feature = "read-tx-timeouts")]
@@ -125,7 +141,10 @@ impl TxnManager {
 
 #[cfg(feature = "read-tx-timeouts")]
 mod read_transactions {
-    use crate::{environment::EnvPtr, error::mdbx_result, txn_manager::TxnManager, Error};
+    use crate::{
+        environment::EnvPtr, error::mdbx_result, metrics::TxnManagerMetrics,
+        txn_manager::TxnManager, Error,
+    };
     use dashmap::{DashMap, DashSet};
     use std::{
         sync::{mpsc::sync_channel, Arc},
@@ -169,6 +188,11 @@ mod read_transactions {
             self.read_transactions.as_ref()?.remove_active(ptr)
         }
 
+        /// Returns `Some(true)` if a transaction has already been aborted due to timeout.
+        pub(crate) fn is_aborted_read_transaction(&self, ptr: *mut ffi::MDBX_txn) -> Option<bool> {
+            Some(self.read_transactions.as_ref()?.is_aborted(ptr))
+        }
+
         /// Removes a transaction from the list of aborted read transactions.
         pub(crate) fn remove_aborted_read_transaction(
             &self,
@@ -188,12 +212,14 @@ mod read_transactions {
         /// We store `usize` instead of a raw pointer as a key, because pointers are not
         /// comparable. The time of transaction opening is stored as a value.
         active: DashMap<usize, Instant>,
-        /// List of read transactions aborted by the [ReadTransactions::start_monitor].
-        /// We keep them until user tries to abort the transaction, so we're able to report a nice
+        /// List of read transactions aborted by the [ReadTransactions::start_monitor]. We keep
+        /// them until the user tries to reuse, so we're able to report a nice
         /// [Error::ReadTransactionAborted] error.
         ///
         /// We store `usize` instead of a raw pointer, because pointers are not comparable.
         aborted: DashSet<usize>,
+        #[doc(hidden)]
+        metrics: TxnManagerMetrics,
     }
 
     impl ReadTransactions {
@@ -211,9 +237,15 @@ mod read_transactions {
             self.active.remove(&(ptr as usize))
         }
 
-        /// Adds a new transaction to the list of aborted read transactions.
-        pub(super) fn add_aborted(&self, ptr: *mut ffi::MDBX_txn) {
-            self.aborted.insert(ptr as usize);
+        /// Adds a new transaction to the set of aborted read transactions. Returns `true` if the
+        /// transaction isn't already in the set.
+        fn add_aborted(&self, ptr: *mut ffi::MDBX_txn) -> bool {
+            self.aborted.insert(ptr as usize)
+        }
+
+        /// Returns `true` if the transaction is in set of aborted transactions.
+        pub(super) fn is_aborted(&self, ptr: *mut ffi::MDBX_txn) -> bool {
+            self.aborted.contains(&(ptr as usize))
         }
 
         /// Removes a transaction from the list of aborted read transactions.
@@ -234,6 +266,7 @@ mod read_transactions {
 
                 loop {
                     let now = Instant::now();
+
                     let mut max_active_transaction_duration = None;
 
                     // Iterate through active read transactions and abort those that's open for
@@ -247,13 +280,13 @@ mod read_transactions {
 
                             // Add the transaction to the list of aborted transactions, so further
                             // usages report the correct error when the transaction is closed.
-                            self.add_aborted(ptr);
+                            _ = self.add_aborted(ptr);
 
-                            // Abort the transaction
                             let result = mdbx_result(unsafe { ffi::mdbx_txn_abort(ptr) });
 
-                            // Add the transaction to `aborted_active`. We can't remove it instantly
-                            // from the list of active transactions, because we iterate through it.
+                            // Add the transaction to `aborted_active`. We can't remove it
+                            // instantly from the list of active transactions, because we iterate
+                            // through it.
                             aborted_active.push((ptr, duration, result.err()));
                         } else {
                             max_active_transaction_duration = Some(
@@ -268,13 +301,9 @@ mod read_transactions {
                         // Try deleting the transaction from the list of active transactions.
                         let was_in_active = self.remove_active(ptr).is_some();
                         if let Some(err) = err {
-                            // If there was an error when aborting the transaction, we need to
-                            // remove it from the list of aborted transactions, because otherwise it
-                            // will stay there forever.
-                            self.remove_aborted(ptr);
                             if was_in_active && err != Error::BadSignature {
-                                // If the transaction was in the list of active transactions and the
-                                // error code is not `EBADSIGN`, then user didn't abort it.
+                                // If the transaction was in the list of active transactions and
+                                // the user didn't abort it.
                                 error!(target: "libmdbx", %err, ?open_duration, "Failed to abort the long-lived read transactions");
                             }
                         } else {
@@ -300,6 +329,8 @@ mod read_transactions {
                         );
                     }
 
+                    self.update_monitor_metrics();
+
                     // Sleep not more than `READ_TRANSACTIONS_CHECK_INTERVAL`, but at least until
                     // the closest deadline of an active read transaction
                     let duration_until_closest_deadline =
@@ -310,6 +341,10 @@ mod read_transactions {
                 }
             });
         }
+
+        fn update_monitor_metrics(&self) {
+            self.metrics.aborted.set(self.aborted.len() as f64)
+        }
     }
 
     #[cfg(test)]
@@ -452,5 +487,33 @@ mod read_transactions {
                 assert!(!read_transactions.aborted.contains(&tx_ptr));
             }
         }
+
+        #[test]
+        fn txn_manager_abort_timed_out_transaction() {
+            const MAX_DURATION: Duration = Duration::from_secs(1);
+
+            let dir = tempdir().unwrap();
+            let env = Environment::builder()
+                .set_max_read_transaction_duration(MaxReadTransactionDuration::Set(MAX_DURATION))
+                .open(dir.path())
+                .unwrap();
+
+            assert!(env.txn_manager().read_transactions.is_some());
+
+            // Create a ro transaction. Abort it by letting it time out.
+            let tx = env.begin_ro_txn().unwrap();
+            let tx_ptr = TxnPtr(tx.txn());
+
+            sleep(MAX_DURATION + READ_TRANSACTIONS_CHECK_INTERVAL);
+
+            // Attempt to abort again throws error.
+            let (sender, rx) = sync_channel(0);
+            env.txn_manager().send_message(TxnManagerMessage::Abort {
+                tx: tx_ptr,
+                flags: RO::OPEN_FLAGS,
+                sender,
+            });
+            assert_eq!(Err(Error::ReadTransactionAborted), rx.recv().unwrap());
+        }
     }
 }