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

reth_db/implementation/mdbx/
tx.rs
