reth_db/implementation/mdbx/
tx.rs

1//! Transaction wrapper for libmdbx-sys.
2
3use super::{cursor::Cursor, utils::*};
4use crate::{
5    metrics::{DatabaseEnvMetrics, Operation, TransactionMode, TransactionOutcome},
6    DatabaseError,
7};
8use reth_db_api::{
9    table::{Compress, DupSort, Encode, IntoVec, Table, TableImporter},
10    transaction::{DbTx, DbTxMut},
11};
12use reth_libmdbx::{ffi::MDBX_dbi, CommitLatency, Transaction, TransactionKind, WriteFlags, RW};
13use reth_primitives_traits::FastInstant as Instant;
14use reth_storage_errors::db::{DatabaseWriteError, DatabaseWriteOperation};
15use reth_tracing::tracing::{debug, instrument, trace, warn};
16use rustc_hash::FxHashMap;
17use std::{
18    backtrace::Backtrace,
19    marker::PhantomData,
20    sync::{
21        atomic::{AtomicBool, Ordering},
22        Arc,
23    },
24    time::Duration,
25};
26
27/// Duration after which we emit the log about long-lived database transactions.
28const LONG_TRANSACTION_DURATION: Duration = Duration::from_secs(60);
29
30/// Wrapper for the libmdbx transaction.
31#[derive(Debug)]
32pub struct Tx<K: TransactionKind> {
33    /// Libmdbx-sys transaction.
34    inner: Transaction<K>,
35
36    /// Cached MDBX DBIs for reuse.
37    dbis: Arc<FxHashMap<&'static str, MDBX_dbi>>,
38
39    /// Handler for metrics with its own [Drop] implementation for cases when the transaction isn't
40    /// closed by [`Tx::commit`] or [`Tx::abort`], but we still need to report it in the metrics.
41    ///
42    /// If [Some], then metrics are reported.
43    metrics_handler: Option<MetricsHandler<K>>,
44}
45
46impl<K: TransactionKind> Tx<K> {
47    /// Creates new `Tx` object with a `RO` or `RW` transaction and optionally enables metrics.
48    #[inline]
49    #[track_caller]
50    pub(crate) fn new(
51        inner: Transaction<K>,
52        dbis: Arc<FxHashMap<&'static str, MDBX_dbi>>,
53        env_metrics: Option<Arc<DatabaseEnvMetrics>>,
54    ) -> reth_libmdbx::Result<Self> {
55        let metrics_handler = env_metrics
56            .map(|env_metrics| {
57                let handler = MetricsHandler::<K>::new(inner.id()?, env_metrics);
58                handler.env_metrics.record_opened_transaction(handler.transaction_mode());
59                handler.log_transaction_opened();
60                Ok(handler)
61            })
62            .transpose()?;
63        Ok(Self { inner, dbis, metrics_handler })
64    }
65
66    /// Returns a reference to the inner libmdbx transaction.
67    pub const fn inner(&self) -> &Transaction<K> {
68        &self.inner
69    }
70
71    /// Gets this transaction ID.
72    pub fn id(&self) -> reth_libmdbx::Result<u64> {
73        self.metrics_handler.as_ref().map_or_else(|| self.inner.id(), |handler| Ok(handler.txn_id))
74    }
75
76    /// Gets a table database handle by name if it exists, otherwise, check the
77    /// database, opening the DB if it exists.
78    pub fn get_dbi_raw(&self, name: &str) -> Result<MDBX_dbi, DatabaseError> {
79        if let Some(dbi) = self.dbis.get(name) {
80            Ok(*dbi)
81        } else {
82            self.inner
83                .open_db(Some(name))
84                .map(|db| db.dbi())
85                .map_err(|e| DatabaseError::Open(e.into()))
86        }
87    }
88
89    /// Gets a table database handle by name if it exists, otherwise, check the
90    /// database, opening the DB if it exists.
91    pub fn get_dbi<T: Table>(&self) -> Result<MDBX_dbi, DatabaseError> {
92        self.get_dbi_raw(T::NAME)
93    }
94
95    /// Create db Cursor
96    pub fn new_cursor<T: Table>(&self) -> Result<Cursor<K, T>, DatabaseError> {
97        let inner = self
98            .inner
99            .cursor_with_dbi(self.get_dbi::<T>()?)
100            .map_err(|e| DatabaseError::InitCursor(e.into()))?;
101
102        Ok(Cursor::new_with_metrics(
103            inner,
104            self.metrics_handler.as_ref().map(|h| h.env_metrics.table_operation_metrics(T::NAME)),
105        ))
106    }
107
108    /// If `self.metrics_handler == Some(_)`, measure the time it takes to execute the closure and
109    /// record a metric with the provided transaction outcome.
110    ///
111    /// Otherwise, just execute the closure.
112    fn execute_with_close_transaction_metric<R>(
113        mut self,
114        outcome: TransactionOutcome,
115        f: impl FnOnce(Self) -> (R, Option<CommitLatency>),
116    ) -> R {
117        let run = |tx| {
118            let start = Instant::now();
119            let (result, commit_latency) = f(tx);
120            let total_duration = start.elapsed();
121
122            if outcome.is_commit() {
123                debug!(
124                    target: "storage::db::mdbx",
125                    ?total_duration,
126                    ?commit_latency,
127                    is_read_only = K::IS_READ_ONLY,
128                    "Commit"
129                );
130            }
131
132            (result, commit_latency, total_duration)
133        };
134
135        if let Some(mut metrics_handler) = self.metrics_handler.take() {
136            metrics_handler.close_recorded = true;
137            metrics_handler.log_backtrace_on_long_read_transaction();
138
139            let (result, commit_latency, close_duration) = run(self);
140            let open_duration = metrics_handler.start.elapsed();
141            metrics_handler.env_metrics.record_closed_transaction(
142                metrics_handler.transaction_mode(),
143                outcome,
144                open_duration,
145                Some(close_duration),
146                commit_latency,
147            );
148
149            result
150        } else {
151            run(self).0
152        }
153    }
154
155    /// If `self.metrics_handler == Some(_)`, measure the time it takes to execute the closure and
156    /// record a metric with the provided operation.
157    ///
158    /// Otherwise, just execute the closure.
159    fn execute_with_operation_metric<T: Table, R>(
160        &self,
161        operation: Operation,
162        value_size: Option<usize>,
163        f: impl FnOnce(&Transaction<K>) -> R,
164    ) -> R {
165        if let Some(metrics_handler) = &self.metrics_handler {
166            metrics_handler.log_backtrace_on_long_read_transaction();
167            metrics_handler
168                .env_metrics
169                .record_operation(T::NAME, operation, value_size, || f(&self.inner))
170        } else {
171            f(&self.inner)
172        }
173    }
174}
175
176#[derive(Debug)]
177struct MetricsHandler<K: TransactionKind> {
178    /// Cached internal transaction ID provided by libmdbx.
179    txn_id: u64,
180    /// The time when transaction has started.
181    ///
182    /// This is a TSC-backed [`reth_primitives_traits::FastInstant`] rather than
183    /// [`std::time::Instant`] because
184    /// [`MetricsHandler::log_backtrace_on_long_read_transaction`] reads it on every database
185    /// operation.
186    start: Instant,
187    /// Duration after which we emit the log about long-lived database transactions.
188    long_transaction_duration: Duration,
189    /// If `true`, the metric about transaction closing has already been recorded and we don't need
190    /// to do anything on [`Drop::drop`].
191    close_recorded: bool,
192    /// If `true`, the backtrace of transaction will be recorded and logged.
193    /// See [`MetricsHandler::log_backtrace_on_long_read_transaction`].
194    record_backtrace: bool,
195    /// If `true`, the backtrace of transaction has already been recorded and logged.
196    /// See [`MetricsHandler::log_backtrace_on_long_read_transaction`].
197    backtrace_recorded: AtomicBool,
198    /// Shared database environment metrics.
199    env_metrics: Arc<DatabaseEnvMetrics>,
200    /// Backtrace of the location where the transaction has been opened. Reported only with debug
201    /// assertions, because capturing the backtrace on every transaction opening is expensive.
202    #[cfg(debug_assertions)]
203    open_backtrace: Backtrace,
204    _marker: PhantomData<K>,
205}
206
207impl<K: TransactionKind> MetricsHandler<K> {
208    fn new(txn_id: u64, env_metrics: Arc<DatabaseEnvMetrics>) -> Self {
209        Self {
210            txn_id,
211            start: Instant::now(),
212            long_transaction_duration: LONG_TRANSACTION_DURATION,
213            close_recorded: false,
214            record_backtrace: true,
215            backtrace_recorded: AtomicBool::new(false),
216            #[cfg(debug_assertions)]
217            open_backtrace: Backtrace::force_capture(),
218            env_metrics,
219            _marker: PhantomData,
220        }
221    }
222
223    const fn transaction_mode(&self) -> TransactionMode {
224        if K::IS_READ_ONLY {
225            TransactionMode::ReadOnly
226        } else {
227            TransactionMode::ReadWrite
228        }
229    }
230
231    /// Logs the caller location and ID of the transaction that was opened.
232    #[track_caller]
233    fn log_transaction_opened(&self) {
234        trace!(
235            target: "storage::db::mdbx",
236            caller = %core::panic::Location::caller(),
237            id = %self.txn_id,
238            mode = %self.transaction_mode().as_str(),
239            "Transaction opened",
240        );
241    }
242
243    /// Logs the backtrace of current call if the duration that the read transaction has been open
244    /// is more than [`LONG_TRANSACTION_DURATION`] and `record_backtrace == true`.
245    /// The backtrace is recorded and logged just once, guaranteed by `backtrace_recorded` atomic.
246    ///
247    /// NOTE: Backtrace is recorded using [`Backtrace::force_capture`], so `RUST_BACKTRACE` env var
248    /// is not needed.
249    fn log_backtrace_on_long_read_transaction(&self) {
250        if self.record_backtrace &&
251            !self.backtrace_recorded.load(Ordering::Relaxed) &&
252            self.transaction_mode().is_read_only()
253        {
254            let open_duration = self.start.elapsed();
255            if open_duration >= self.long_transaction_duration {
256                self.backtrace_recorded.store(true, Ordering::Relaxed);
257                #[cfg(debug_assertions)]
258                let open_backtrace = format_args!("{}", self.open_backtrace);
259                #[cfg(not(debug_assertions))]
260                let open_backtrace = tracing::field::Empty;
261                warn!(
262                    target: "storage::db::mdbx",
263                    ?open_duration,
264                    id=%self.txn_id,
265                    backtrace=%Backtrace::force_capture(),
266                    open_backtrace,
267                    "A database read transaction has been open for too long"
268                );
269            }
270        }
271    }
272}
273
274impl<K: TransactionKind> Drop for MetricsHandler<K> {
275    fn drop(&mut self) {
276        if !self.close_recorded {
277            self.log_backtrace_on_long_read_transaction();
278            self.env_metrics.record_closed_transaction(
279                self.transaction_mode(),
280                TransactionOutcome::Drop,
281                self.start.elapsed(),
282                None,
283                None,
284            );
285        }
286    }
287}
288
289impl TableImporter for Tx<RW> {}
290
291impl<K: TransactionKind> DbTx for Tx<K> {
292    type Cursor<T: Table> = Cursor<K, T>;
293    type DupCursor<T: DupSort> = Cursor<K, T>;
294
295    fn get<T: Table>(&self, key: T::Key) -> Result<Option<<T as Table>::Value>, DatabaseError> {
296        self.get_by_encoded_key::<T>(&key.encode())
297    }
298
299    fn get_by_encoded_key<T: Table>(
300        &self,
301        key: &<T::Key as Encode>::Encoded,
302    ) -> Result<Option<T::Value>, DatabaseError> {
303        self.execute_with_operation_metric::<T, _>(Operation::Get, None, |tx| {
304            tx.get(self.get_dbi::<T>()?, key.as_ref())
305                .map_err(|e| DatabaseError::Read(e.into()))?
306                .map(decode_one::<T>)
307                .transpose()
308        })
309    }
310
311    #[instrument(name = "Tx::commit", level = "debug", target = "providers::db", skip_all)]
312    fn commit(self) -> Result<(), DatabaseError> {
313        self.execute_with_close_transaction_metric(TransactionOutcome::Commit, |this| {
314            match this.inner.commit().map_err(|e| DatabaseError::Commit(e.into())) {
315                Ok(latency) => (Ok(()), Some(latency)),
316                Err(e) => (Err(e), None),
317            }
318        })
319    }
320
321    fn abort(self) {
322        self.execute_with_close_transaction_metric(TransactionOutcome::Abort, |this| {
323            (drop(this.inner), None)
324        })
325    }
326
327    // Iterate over read only values in database.
328    fn cursor_read<T: Table>(&self) -> Result<Self::Cursor<T>, DatabaseError> {
329        self.new_cursor()
330    }
331
332    /// Iterate over read only values in database.
333    fn cursor_dup_read<T: DupSort>(&self) -> Result<Self::DupCursor<T>, DatabaseError> {
334        self.new_cursor()
335    }
336
337    /// Returns number of entries in the table using cheap DB stats invocation.
338    fn entries<T: Table>(&self) -> Result<usize, DatabaseError> {
339        Ok(self
340            .inner
341            .db_stat_with_dbi(self.get_dbi::<T>()?)
342            .map_err(|e| DatabaseError::Stats(e.into()))?
343            .entries())
344    }
345
346    /// Disables long-lived read transaction safety guarantees, such as backtrace recording and
347    /// timeout.
348    fn disable_long_read_transaction_safety(&mut self) {
349        if let Some(metrics_handler) = self.metrics_handler.as_mut() {
350            metrics_handler.record_backtrace = false;
351        }
352
353        self.inner.disable_timeout();
354    }
355}
356
357#[derive(Clone, Copy)]
358enum PutKind {
359    /// Default kind that inserts a new key-value or overwrites an existed key.
360    Upsert,
361    /// Append the key-value to the end of the table -- fast path when the new
362    /// key is the highest so far, like the latest block number.
363    Append,
364}
365
366impl PutKind {
367    const fn into_operation_and_flags(self) -> (Operation, DatabaseWriteOperation, WriteFlags) {
368        match self {
369            Self::Upsert => {
370                (Operation::PutUpsert, DatabaseWriteOperation::PutUpsert, WriteFlags::UPSERT)
371            }
372            Self::Append => {
373                (Operation::PutAppend, DatabaseWriteOperation::PutAppend, WriteFlags::APPEND)
374            }
375        }
376    }
377}
378
379impl Tx<RW> {
380    /// The inner implementation mapping to `mdbx_put` that supports different
381    /// put kinds like upserting and appending.
382    fn put<T: Table>(
383        &self,
384        kind: PutKind,
385        key: T::Key,
386        value: T::Value,
387    ) -> Result<(), DatabaseError> {
388        let key = key.encode();
389        let value = value.compress();
390        let (operation, write_operation, flags) = kind.into_operation_and_flags();
391        self.execute_with_operation_metric::<T, _>(operation, Some(value.as_ref().len()), |tx| {
392            tx.put(self.get_dbi::<T>()?, key.as_ref(), value, flags).map_err(|e| {
393                DatabaseWriteError {
394                    info: e.into(),
395                    operation: write_operation,
396                    table_name: T::NAME,
397                    key: key.into_vec(),
398                }
399                .into()
400            })
401        })
402    }
403}
404
405impl DbTxMut for Tx<RW> {
406    type CursorMut<T: Table> = Cursor<RW, T>;
407    type DupCursorMut<T: DupSort> = Cursor<RW, T>;
408
409    fn put<T: Table>(&self, key: T::Key, value: T::Value) -> Result<(), DatabaseError> {
410        self.put::<T>(PutKind::Upsert, key, value)
411    }
412
413    fn append<T: Table>(&self, key: T::Key, value: T::Value) -> Result<(), DatabaseError> {
414        self.put::<T>(PutKind::Append, key, value)
415    }
416
417    fn delete<T: Table>(
418        &self,
419        key: T::Key,
420        value: Option<T::Value>,
421    ) -> Result<bool, DatabaseError> {
422        let mut data = None;
423
424        let value = value.map(Compress::compress);
425        if let Some(value) = &value {
426            data = Some(value.as_ref());
427        };
428
429        self.execute_with_operation_metric::<T, _>(Operation::Delete, None, |tx| {
430            tx.del(self.get_dbi::<T>()?, key.encode(), data)
431                .map_err(|e| DatabaseError::Delete(e.into()))
432        })
433    }
434
435    fn clear<T: Table>(&self) -> Result<(), DatabaseError> {
436        self.inner.clear_db(self.get_dbi::<T>()?).map_err(|e| DatabaseError::Delete(e.into()))?;
437
438        Ok(())
439    }
440
441    fn cursor_write<T: Table>(&self) -> Result<Self::CursorMut<T>, DatabaseError> {
442        self.new_cursor()
443    }
444
445    fn cursor_dup_write<T: DupSort>(&self) -> Result<Self::DupCursorMut<T>, DatabaseError> {
446        self.new_cursor()
447    }
448}
449
450#[cfg(test)]
451mod tests {
452    use crate::{mdbx::DatabaseArguments, tables, DatabaseEnv, DatabaseEnvKind};
453    use reth_db_api::{database::Database, models::ClientVersion, transaction::DbTx};
454    use reth_libmdbx::MaxReadTransactionDuration;
455    use reth_storage_errors::db::DatabaseError;
456    use std::{sync::atomic::Ordering, thread::sleep, time::Duration};
457    use tempfile::tempdir;
458
459    #[test]
460    fn long_read_transaction_safety_disabled() {
461        const MAX_DURATION: Duration = Duration::from_secs(1);
462
463        let dir = tempdir().unwrap();
464        let args = DatabaseArguments::new(ClientVersion::default())
465            .with_max_read_transaction_duration(Some(MaxReadTransactionDuration::Set(
466                MAX_DURATION,
467            )));
468        let db = DatabaseEnv::open(dir.path(), DatabaseEnvKind::RW, args).unwrap().with_metrics();
469
470        let mut tx = db.tx().unwrap();
471        tx.metrics_handler.as_mut().unwrap().long_transaction_duration = MAX_DURATION;
472        tx.disable_long_read_transaction_safety();
473        // Give the `TxnManager` some time to time out the transaction.
474        sleep(MAX_DURATION + Duration::from_millis(100));
475
476        // Transaction has not timed out.
477        assert!(matches!(
478            tx.get::<tables::Transactions>(0).unwrap_err(),
479            DatabaseError::Open(err) if err == reth_libmdbx::Error::NotFound.into()));
480        // Backtrace is not recorded.
481        assert!(!tx.metrics_handler.unwrap().backtrace_recorded.load(Ordering::Relaxed));
482    }
483
484    #[test]
485    fn long_read_transaction_safety_enabled() {
486        const MAX_DURATION: Duration = Duration::from_secs(1);
487
488        let dir = tempdir().unwrap();
489        let args = DatabaseArguments::new(ClientVersion::default())
490            .with_max_read_transaction_duration(Some(MaxReadTransactionDuration::Set(
491                MAX_DURATION,
492            )));
493        let db = DatabaseEnv::open(dir.path(), DatabaseEnvKind::RW, args).unwrap().with_metrics();
494
495        let mut tx = db.tx().unwrap();
496        tx.metrics_handler.as_mut().unwrap().long_transaction_duration = MAX_DURATION;
497        // Give the `TxnManager` some time to time out the transaction.
498        sleep(MAX_DURATION + Duration::from_millis(100));
499
500        // Transaction has timed out.
501        assert!(matches!(
502            tx.get::<tables::Transactions>(0).unwrap_err(),
503            DatabaseError::Open(err) if err == reth_libmdbx::Error::ReadTransactionTimeout.into()));
504        // Backtrace is recorded.
505        assert!(tx.metrics_handler.unwrap().backtrace_recorded.load(Ordering::Relaxed));
506    }
507}
reth_db/implementation/mdbx/tx.rs

reth_db/implementation/mdbx/
tx.rs
