foundationdb 0.10.0

// Copyright 2018 foundationdb-rs developers, https://github.com/Clikengo/foundationdb-rs/graphs/contributors
// Copyright 2013-2018 Apple, Inc and the FoundationDB project authors.
//
// Licensed under the Apache License, Version 2.0, <LICENSE-APACHE or
// http://apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
// http://opensource.org/licenses/MIT>, at your option. This file may not be
// copied, modified, or distributed except according to those terms.

//! Implementations of the FDBTransaction C API
//!
//! <https://apple.github.io/foundationdb/api-c.html#transaction>

use foundationdb_sys as fdb_sys;
use std::fmt;
use std::ops::{Deref, Range, RangeInclusive};
use std::ptr::NonNull;
use std::sync::Arc;

use crate::future::*;
use crate::keyselector::*;
use crate::metrics::{FdbCommand, TransactionMetrics};
use crate::options;

use crate::{error, FdbError, FdbResult};
use foundationdb_macros::cfg_api_versions;

use crate::error::{FdbBindingError, TransactionMetricsNotFound};

use futures::{
    future, future::Either, stream, Future, FutureExt, Stream, TryFutureExt, TryStreamExt,
};

#[cfg_api_versions(min = 610)]
const METADATA_VERSION_KEY: &[u8] = b"\xff/metadataVersion";

/// A committed transaction.
#[derive(Debug)]
#[repr(transparent)]
pub struct TransactionCommitted {
    tr: Transaction,
}

impl TransactionCommitted {
    /// Retrieves the database version number at which a given transaction was committed.
    ///
    /// Read-only transactions do not modify the database when committed and will have a committed
    /// version of -1. Keep in mind that a transaction which reads keys and then sets them to their
    /// current values may be optimized to a read-only transaction.
    ///
    /// Note that database versions are not necessarily unique to a given transaction and so cannot
    /// be used to determine in what order two transactions completed. The only use for this
    /// function is to manually enforce causal consistency when calling `set_read_version()` on
    /// another subsequent transaction.
    ///
    /// Most applications will not call this function.
    pub fn committed_version(&self) -> FdbResult<i64> {
        let mut version: i64 = 0;
        error::eval(unsafe {
            fdb_sys::fdb_transaction_get_committed_version(self.tr.inner.as_ptr(), &mut version)
        })?;
        Ok(version)
    }

    /// Reset the transaction to its initial state.
    ///
    /// This will not affect previously committed data.
    ///
    /// This is similar to dropping the transaction and creating a new one.
    pub fn reset(mut self) -> Transaction {
        self.tr.reset();
        self.tr
    }
}
impl From<TransactionCommitted> for Transaction {
    fn from(tc: TransactionCommitted) -> Transaction {
        tc.reset()
    }
}

/// A failed to commit transaction.
pub struct TransactionCommitError {
    tr: Transaction,
    err: FdbError,
}

impl TransactionCommitError {
    /// Implements the recommended retry and backoff behavior for a transaction. This function knows
    /// which of the error codes generated by other `Transaction` functions represent temporary
    /// error conditions and which represent application errors that should be handled by the
    /// application. It also implements an exponential backoff strategy to avoid swamping the
    /// database cluster with excessive retries when there is a high level of conflict between
    /// transactions.
    ///
    /// You should not call this method most of the times and use `Database::transact` which
    /// implements a retry loop strategy for you.
    pub fn on_error(self) -> impl Future<Output = FdbResult<Transaction>> {
        FdbFuture::<()>::new(unsafe {
            fdb_sys::fdb_transaction_on_error(self.tr.inner.as_ptr(), self.err.code())
        })
        .map_ok(|()| self.tr)
    }

    /// Reset the transaction to its initial state.
    ///
    /// This is similar to dropping the transaction and creating a new one.
    pub fn reset(mut self) -> Transaction {
        self.tr.reset();
        self.tr
    }
}

impl Deref for TransactionCommitError {
    type Target = FdbError;
    fn deref(&self) -> &FdbError {
        &self.err
    }
}

impl From<TransactionCommitError> for FdbError {
    fn from(tce: TransactionCommitError) -> FdbError {
        tce.err
    }
}

impl fmt::Debug for TransactionCommitError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "TransactionCommitError({})", self.err)
    }
}

impl fmt::Display for TransactionCommitError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        self.err.fmt(f)
    }
}

impl std::error::Error for TransactionCommitError {}

/// The result of `Transaction::Commit`
type TransactionResult = Result<TransactionCommitted, TransactionCommitError>;

/// A cancelled transaction
#[derive(Debug)]
#[repr(transparent)]
pub struct TransactionCancelled {
    tr: Transaction,
}
impl TransactionCancelled {
    /// Reset the transaction to its initial state.
    ///
    /// This is similar to dropping the transaction and creating a new one.
    pub fn reset(mut self) -> Transaction {
        self.tr.reset();
        self.tr
    }
}
impl From<TransactionCancelled> for Transaction {
    fn from(tc: TransactionCancelled) -> Transaction {
        tc.reset()
    }
}

/// In FoundationDB, a transaction is a mutable snapshot of a database.
///
/// All read and write operations on a transaction see and modify an otherwise-unchanging version of the database and only change the underlying database if and when the transaction is committed. Read operations do see the effects of previous write operations on the same transaction. Committing a transaction usually succeeds in the absence of conflicts.
///
/// Applications must provide error handling and an appropriate retry loop around the application code for a transaction. See the documentation for [fdb_transaction_on_error()](https://apple.github.io/foundationdb/api-c.html#transaction).
///
/// Transactions group operations into a unit with the properties of atomicity, isolation, and durability. Transactions also provide the ability to maintain an application’s invariants or integrity constraints, supporting the property of consistency. Together these properties are known as ACID.
///
/// Transactions are also causally consistent: once a transaction has been successfully committed, all subsequently created transactions will see the modifications made by it.
#[derive(Debug)]
pub struct Transaction {
    // Order of fields should not be changed, because Rust drops field top-to-bottom, and
    // transaction should be dropped before cluster.
    inner: NonNull<fdb_sys::FDBTransaction>,
    metrics: Option<TransactionMetrics>,
}
unsafe impl Send for Transaction {}
unsafe impl Sync for Transaction {}

/// Converts Rust `bool` into `fdb_sys::fdb_bool_t`
#[inline]
fn fdb_bool(v: bool) -> fdb_sys::fdb_bool_t {
    if v {
        1
    } else {
        0
    }
}
#[inline]
fn fdb_len(len: usize, context: &'static str) -> std::os::raw::c_int {
    assert!(len <= i32::MAX as usize, "{context}.len() > i32::MAX");
    len as i32
}
#[inline]
fn fdb_iteration(iteration: usize) -> std::os::raw::c_int {
    if iteration > i32::MAX as usize {
        0 // this will cause client_invalid_operation
    } else {
        iteration as i32
    }
}
#[inline]
fn fdb_limit(v: usize) -> std::os::raw::c_int {
    if v > i32::MAX as usize {
        i32::MAX
    } else {
        v as i32
    }
}

/// `RangeOption` represents a query parameters for range scan query.
///
/// You can construct `RangeOption` easily:
///
/// ```
/// use foundationdb::RangeOption;
///
/// let opt = RangeOption::from((b"begin".as_ref(), b"end".as_ref()));
/// let opt: RangeOption = (b"begin".as_ref()..b"end".as_ref()).into();
/// let opt = RangeOption {
///     limit: Some(10),
///     ..RangeOption::from((b"begin".as_ref(), b"end".as_ref()))
/// };
/// ```
#[derive(Debug, Clone)]
pub struct RangeOption<'a> {
    /// The beginning of the range.
    pub begin: KeySelector<'a>,
    /// The end of the range.
    pub end: KeySelector<'a>,
    /// If non-zero, indicates the maximum number of key-value pairs to return.
    pub limit: Option<usize>,
    /// If non-zero, indicates a (soft) cap on the combined number of bytes of keys and values to
    /// return for each item.
    pub target_bytes: usize,
    /// One of the options::StreamingMode values indicating how the caller would like the data in
    /// the range returned.
    pub mode: options::StreamingMode,
    /// If true, key-value pairs will be returned in reverse lexicographical order beginning at
    /// the end of the range.
    pub reverse: bool,
    #[doc(hidden)]
    pub __non_exhaustive: std::marker::PhantomData<()>,
}

impl RangeOption<'_> {
    /// Reverses the range direction.
    pub fn rev(mut self) -> Self {
        self.reverse = !self.reverse;
        self
    }

    pub fn next_range(mut self, kvs: &FdbValues) -> Option<Self> {
        if !kvs.more() {
            return None;
        }

        let last = kvs.last()?;
        let last_key = last.key();

        if let Some(limit) = self.limit.as_mut() {
            *limit = limit.saturating_sub(kvs.len());
            if *limit == 0 {
                return None;
            }
        }

        if self.reverse {
            self.end.make_first_greater_or_equal(last_key);
        } else {
            self.begin.make_first_greater_than(last_key);
        }
        Some(self)
    }

    #[cfg_api_versions(min = 710)]
    pub(crate) fn next_mapped_range(mut self, kvs: &MappedKeyValues) -> Option<Self> {
        if !kvs.more() {
            return None;
        }

        let last = kvs.last()?;
        let last_key = last.parent_key();

        if let Some(limit) = self.limit.as_mut() {
            *limit = limit.saturating_sub(kvs.len());
            if *limit == 0 {
                return None;
            }
        }

        if self.reverse {
            self.end.make_first_greater_or_equal(last_key);
        } else {
            self.begin.make_first_greater_than(last_key);
        }
        Some(self)
    }
}

impl Default for RangeOption<'_> {
    fn default() -> Self {
        Self {
            begin: KeySelector::first_greater_or_equal([].as_ref()),
            end: KeySelector::first_greater_or_equal([].as_ref()),
            limit: None,
            target_bytes: 0,
            mode: options::StreamingMode::Iterator,
            reverse: false,
            __non_exhaustive: std::marker::PhantomData,
        }
    }
}

impl<'a> From<(KeySelector<'a>, KeySelector<'a>)> for RangeOption<'a> {
    fn from((begin, end): (KeySelector<'a>, KeySelector<'a>)) -> Self {
        Self {
            begin,
            end,
            ..Self::default()
        }
    }
}
impl From<(Vec<u8>, Vec<u8>)> for RangeOption<'static> {
    fn from((begin, end): (Vec<u8>, Vec<u8>)) -> Self {
        Self {
            begin: KeySelector::first_greater_or_equal(begin),
            end: KeySelector::first_greater_or_equal(end),
            ..Self::default()
        }
    }
}
impl<'a> From<(&'a [u8], &'a [u8])> for RangeOption<'a> {
    fn from((begin, end): (&'a [u8], &'a [u8])) -> Self {
        Self {
            begin: KeySelector::first_greater_or_equal(begin),
            end: KeySelector::first_greater_or_equal(end),
            ..Self::default()
        }
    }
}
impl<'a> From<std::ops::Range<KeySelector<'a>>> for RangeOption<'a> {
    fn from(range: Range<KeySelector<'a>>) -> Self {
        RangeOption::from((range.start, range.end))
    }
}

impl<'a> From<std::ops::Range<&'a [u8]>> for RangeOption<'a> {
    fn from(range: Range<&'a [u8]>) -> Self {
        RangeOption::from((range.start, range.end))
    }
}

impl From<std::ops::Range<std::vec::Vec<u8>>> for RangeOption<'static> {
    fn from(range: Range<Vec<u8>>) -> Self {
        RangeOption::from((range.start, range.end))
    }
}

impl<'a> From<std::ops::RangeInclusive<&'a [u8]>> for RangeOption<'a> {
    fn from(range: RangeInclusive<&'a [u8]>) -> Self {
        let (start, end) = range.into_inner();
        (KeySelector::first_greater_or_equal(start)..KeySelector::first_greater_than(end)).into()
    }
}

impl From<std::ops::RangeInclusive<std::vec::Vec<u8>>> for RangeOption<'static> {
    fn from(range: RangeInclusive<Vec<u8>>) -> Self {
        let (start, end) = range.into_inner();
        (KeySelector::first_greater_or_equal(start)..KeySelector::first_greater_than(end)).into()
    }
}

impl Transaction {
    pub(crate) fn new(inner: NonNull<fdb_sys::FDBTransaction>) -> Self {
        Self {
            inner,
            metrics: None,
        }
    }

    pub fn new_instrumented(
        inner: NonNull<fdb_sys::FDBTransaction>,
        metrics: TransactionMetrics,
    ) -> Self {
        Self {
            inner,
            metrics: Some(metrics),
        }
    }

    /// Called to set an option on an FDBTransaction.
    pub fn set_option(&self, opt: options::TransactionOption) -> FdbResult<()> {
        unsafe { opt.apply(self.inner.as_ptr()) }
    }

    /// Pass through an option given a code and raw data. Useful when creating a passthrough layer
    /// where the code and data will be provided as raw, in order to avoid deserializing to an option
    /// and serializing it back to code and data.
    /// In general, you should use `set_option`.
    pub fn set_raw_option(
        &self,
        code: fdb_sys::FDBTransactionOption,
        data: Option<Vec<u8>>,
    ) -> FdbResult<()> {
        let (data_ptr, size) = data
            .as_ref()
            .map(|data| {
                (
                    data.as_ptr(),
                    i32::try_from(data.len()).expect("len to fit in i32"),
                )
            })
            .unwrap_or_else(|| (std::ptr::null(), 0));
        let err = unsafe {
            fdb_sys::fdb_transaction_set_option(self.inner.as_ptr(), code, data_ptr, size)
        };
        if err != 0 {
            Err(FdbError::from_code(err))
        } else {
            Ok(())
        }
    }

    /// Modify the database snapshot represented by transaction to change the given
    /// key to have the given value.
    ///
    /// If the given key was not previously present in the database it is inserted.
    /// The modification affects the actual database only if transaction is later
    /// committed with `Transaction::commit`.
    ///
    /// # Arguments
    ///
    /// * `key` - the name of the key to be inserted into the database.
    /// * `value` - the value to be inserted into the database
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, key, value))
    )]
    pub fn set(&self, key: &[u8], value: &[u8]) {
        unsafe {
            fdb_sys::fdb_transaction_set(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
                value.as_ptr(),
                fdb_len(value.len(), "value"),
            )
        }

        if let Some(metrics) = &self.metrics {
            metrics.report_metrics(FdbCommand::Set((key.len() + value.len()) as u64));
        }
    }

    /// Modify the database snapshot represented by transaction to remove the given key from the
    /// database.
    ///
    /// If the key was not previously present in the database, there is no effect. The modification
    /// affects the actual database only if transaction is later committed with
    /// `Transaction::commit`.
    ///
    /// # Arguments
    ///
    /// * `key` - the name of the key to be removed from the database.
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, key))
    )]
    pub fn clear(&self, key: &[u8]) {
        unsafe {
            fdb_sys::fdb_transaction_clear(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
            )
        }

        if let Some(metrics) = &self.metrics {
            metrics.report_metrics(FdbCommand::Clear);
        }
    }

    /// Reads a value from the database snapshot represented by transaction.
    ///
    /// Returns an FDBFuture which will be set to the value of key in the database if there is any.
    ///
    /// # Arguments
    ///
    /// * `key` - the name of the key to be looked up in the database
    /// * `snapshot` - `true` if this is a [snapshot read](https://apple.github.io/foundationdb/api-c.html#snapshots)
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, key))
    )]
    pub fn get(
        &self,
        key: &[u8],
        snapshot: bool,
    ) -> impl Future<Output = FdbResult<Option<FdbSlice>>> + Send + Sync + Unpin {
        let metrics = self.metrics.clone();
        let lenght_key = key.len();

        FdbFuture::<Option<FdbSlice>>::new(unsafe {
            fdb_sys::fdb_transaction_get(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
                fdb_bool(snapshot),
            )
        })
        .map(move |result| {
            if let Ok(value) = &result {
                if let Some(metrics) = metrics.as_ref() {
                    let (bytes_count, kv_fetched) = if let Some(values) = value {
                        ((lenght_key + values.len()) as u64, 1)
                    } else {
                        (lenght_key as u64, 0)
                    };
                    metrics.report_metrics(FdbCommand::Get(bytes_count, kv_fetched));
                }
            }
            result
        })
    }

    /// Modify the database snapshot represented by transaction to perform the operation indicated
    /// by operationType with operand param to the value stored by the given key.
    ///
    /// An atomic operation is a single database command that carries out several logical steps:
    /// reading the value of a key, performing a transformation on that value, and writing the
    /// result. Different atomic operations perform different transformations. Like other database
    /// operations, an atomic operation is used within a transaction; however, its use within a
    /// transaction will not cause the transaction to conflict.
    ///
    /// Atomic operations do not expose the current value of the key to the client but simply send
    /// the database the transformation to apply. In regard to conflict checking, an atomic
    /// operation is equivalent to a write without a read. It can only cause other transactions
    /// performing reads of the key to conflict.
    ///
    /// By combining these logical steps into a single, read-free operation, FoundationDB can
    /// guarantee that the transaction will not conflict due to the operation. This makes atomic
    /// operations ideal for operating on keys that are frequently modified. A common example is
    /// the use of a key-value pair as a counter.
    ///
    /// # Warning
    ///
    /// If a transaction uses both an atomic operation and a strictly serializable read on the same
    /// key, the benefits of using the atomic operation (for both conflict checking and performance)
    /// are lost.
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, key, param))
    )]
    pub fn atomic_op(&self, key: &[u8], param: &[u8], op_type: options::MutationType) {
        unsafe {
            fdb_sys::fdb_transaction_atomic_op(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
                param.as_ptr(),
                fdb_len(param.len(), "param"),
                op_type.code(),
            )
        }

        if let Some(metrics) = &self.metrics {
            metrics.report_metrics(FdbCommand::Atomic);
        }
    }

    /// Resolves a key selector against the keys in the database snapshot represented by
    /// transaction.
    ///
    /// Returns an FDBFuture which will be set to the key in the database matching the key
    /// selector.
    ///
    /// # Arguments
    ///
    /// * `selector`: the key selector
    /// * `snapshot`: `true` if this is a [snapshot read](https://apple.github.io/foundationdb/api-c.html#snapshots)
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, selector, snapshot))
    )]
    pub fn get_key(
        &self,
        selector: &KeySelector,
        snapshot: bool,
    ) -> impl Future<Output = FdbResult<FdbSlice>> + Send + Sync + Unpin {
        let key = selector.key();
        FdbFuture::new(unsafe {
            fdb_sys::fdb_transaction_get_key(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
                fdb_bool(selector.or_equal()),
                selector.offset(),
                fdb_bool(snapshot),
            )
        })
    }

    /// Reads all key-value pairs in the database snapshot represented by transaction (potentially
    /// limited by limit, target_bytes, or mode) which have a key lexicographically greater than or
    /// equal to the key resolved by the begin key selector and lexicographically less than the key
    /// resolved by the end key selector.
    ///
    /// Returns a stream of KeyValue slices.
    ///
    /// This method is a little more efficient than `get_ranges_keyvalues` but a little harder to
    /// use.
    ///
    /// # Arguments
    ///
    /// * `opt`: the range, limit, target_bytes and mode
    /// * `snapshot`: `true` if this is a [snapshot read](https://apple.github.io/foundationdb/api-c.html#snapshots)
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, opt, snapshot))
    )]
    pub fn get_ranges<'a>(
        &'a self,
        opt: RangeOption<'a>,
        snapshot: bool,
    ) -> impl Stream<Item = FdbResult<FdbValues>> + Send + Sync + Unpin + 'a {
        stream::unfold((1, Some(opt)), move |(iteration, maybe_opt)| {
            if let Some(opt) = maybe_opt {
                Either::Left(self.get_range(&opt, iteration as usize, snapshot).map(
                    move |maybe_values| {
                        let next_opt = match &maybe_values {
                            Ok(values) => opt.next_range(values),
                            Err(..) => None,
                        };
                        Some((maybe_values, (iteration + 1, next_opt)))
                    },
                ))
            } else {
                Either::Right(future::ready(None))
            }
        })
    }

    /// Reads all key-value pairs in the database snapshot represented by transaction (potentially
    /// limited by limit, target_bytes, or mode) which have a key lexicographically greater than or
    /// equal to the key resolved by the begin key selector and lexicographically less than the key
    /// resolved by the end key selector.
    ///
    /// Returns a stream of KeyValue.
    ///
    /// # Arguments
    ///
    /// * `opt`: the range, limit, target_bytes and mode
    /// * `snapshot`: `true` if this is a [snapshot read](https://apple.github.io/foundationdb/api-c.html#snapshots)
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, opt, snapshot))
    )]
    pub fn get_ranges_keyvalues<'a>(
        &'a self,
        opt: RangeOption<'a>,
        snapshot: bool,
    ) -> impl Stream<Item = FdbResult<FdbValue>> + Unpin + 'a {
        self.get_ranges(opt, snapshot)
            .map_ok(|values| stream::iter(values.into_iter().map(Ok)))
            .try_flatten()
    }

    /// Reads all key-value pairs in the database snapshot represented by transaction (potentially
    /// limited by limit, target_bytes, or mode) which have a key lexicographically greater than or
    /// equal to the key resolved by the begin key selector and lexicographically less than the key
    /// resolved by the end key selector.
    ///
    /// # Arguments
    ///
    /// * `opt`: the range, limit, target_bytes and mode
    /// * `iteration`: If opt.mode is Iterator, this parameter should start at 1 and be incremented
    ///   by 1 for each successive call while reading this range. In all other cases it is ignored.
    /// * `snapshot`: `true` if this is a [snapshot read](https://apple.github.io/foundationdb/api-c.html#snapshots)
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, opt, snapshot))
    )]
    pub fn get_range(
        &self,
        opt: &RangeOption,
        iteration: usize,
        snapshot: bool,
    ) -> impl Future<Output = FdbResult<FdbValues>> + Send + Sync + Unpin {
        let begin = &opt.begin;
        let end = &opt.end;
        let key_begin = begin.key();
        let key_end = end.key();

        let metrics = self.metrics.clone();

        FdbFuture::<FdbValues>::new(unsafe {
            fdb_sys::fdb_transaction_get_range(
                self.inner.as_ptr(),
                key_begin.as_ptr(),
                fdb_len(key_begin.len(), "key_begin"),
                fdb_bool(begin.or_equal()),
                begin.offset(),
                key_end.as_ptr(),
                fdb_len(key_end.len(), "key_end"),
                fdb_bool(end.or_equal()),
                end.offset(),
                fdb_limit(opt.limit.unwrap_or(0)),
                fdb_limit(opt.target_bytes),
                opt.mode.code(),
                fdb_iteration(iteration),
                fdb_bool(snapshot),
                fdb_bool(opt.reverse),
            )
        })
        .map(move |result| {
            if let (Ok(values), Some(metrics)) = (&result, metrics) {
                let kv_fetched = values.len();
                let mut bytes_count = 0;

                for key_value in values.as_ref() {
                    let key_len = key_value.key().len();
                    let value_len = key_value.value().len();

                    bytes_count += (key_len + value_len) as u64
                }

                metrics.report_metrics(FdbCommand::GetRange(bytes_count, kv_fetched as u64));
            };

            result
        })
    }

    /// Mapped Range is an experimental feature introduced in FDB 7.1.
    /// It is intended to improve the client throughput and reduce latency for querying data through a Subspace used as a "index".
    /// In such a case, querying records by scanning an index in relational databases can be
    /// translated to a GetRange request on the index entries followed up by multiple GetValue requests for the record entries in FDB.
    ///
    /// This method is allowing FoundationDB "follow up" a GetRange request with GetValue requests,
    /// this can happen in one request without additional back and forth. Considering the overhead
    /// of each request, this saves time and resources on serialization, deserialization, and network.
    ///
    /// A mapped request will:
    ///
    /// * Do a range query (same as a `Transaction.get_range` request) and get the result. We call it the primary query.
    /// * For each key-value pair in the primary query result, translate it to a `get_range` query and get the result. We call them secondary queries.
    /// * Put all results in a nested structure and return them.
    ///
    /// **WARNING** : This feature is considered experimental at this time. It is only allowed when
    /// using snapshot isolation AND disabling read-your-writes.
    ///
    /// More info can be found in the relevant [documentation](https://github.com/apple/foundationdb/wiki/Everything-about-GetMappedRange#input).
    ///
    /// This is the "raw" version, users are expected to use [Transaction::get_mapped_ranges]
    #[cfg_api_versions(min = 710)]
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, opt, mapper, snapshot))
    )]
    pub fn get_mapped_range(
        &self,
        opt: &RangeOption,
        mapper: &[u8],
        iteration: usize,
        snapshot: bool,
    ) -> impl Future<Output = FdbResult<MappedKeyValues>> + Send + Sync + Unpin {
        let begin = &opt.begin;
        let end = &opt.end;
        let key_begin = begin.key();
        let key_end = end.key();

        FdbFuture::new(unsafe {
            fdb_sys::fdb_transaction_get_mapped_range(
                self.inner.as_ptr(),
                key_begin.as_ptr(),
                fdb_len(key_begin.len(), "key_begin"),
                fdb_bool(begin.or_equal()),
                begin.offset(),
                key_end.as_ptr(),
                fdb_len(key_end.len(), "key_end"),
                fdb_bool(end.or_equal()),
                end.offset(),
                mapper.as_ptr(),
                fdb_len(mapper.len(), "mapper_length"),
                fdb_limit(opt.limit.unwrap_or(0)),
                fdb_limit(opt.target_bytes),
                opt.mode.code(),
                fdb_iteration(iteration),
                fdb_bool(snapshot),
                fdb_bool(opt.reverse),
            )
        })
    }

    /// Mapped Range is an experimental feature introduced in FDB 7.1.
    /// It is intended to improve the client throughput and reduce latency for querying data through a Subspace used as a "index".
    /// In such a case, querying records by scanning an index in relational databases can be
    /// translated to a GetRange request on the index entries followed up by multiple GetValue requests for the record entries in FDB.
    ///
    /// This method is allowing FoundationDB "follow up" a GetRange request with GetValue requests,
    /// this can happen in one request without additional back and forth. Considering the overhead
    /// of each request, this saves time and resources on serialization, deserialization, and network.
    ///
    /// A mapped request will:
    ///
    /// * Do a range query (same as a `Transaction.get_range` request) and get the result. We call it the primary query.
    /// * For each key-value pair in the primary query result, translate it to a `get_range` query and get the result. We call them secondary queries.
    /// * Put all results in a nested structure and return them.
    ///
    /// **WARNING** : This feature is considered experimental at this time. It is only allowed when
    /// using snapshot isolation AND disabling read-your-writes.
    ///
    /// More info can be found in the relevant [documentation](https://github.com/apple/foundationdb/wiki/Everything-about-GetMappedRange#input).
    #[cfg_api_versions(min = 710)]
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, opt, mapper, snapshot))
    )]
    pub fn get_mapped_ranges<'a>(
        &'a self,
        opt: RangeOption<'a>,
        mapper: &'a [u8],
        snapshot: bool,
    ) -> impl Stream<Item = FdbResult<MappedKeyValues>> + Send + Sync + Unpin + 'a {
        stream::unfold((1, Some(opt)), move |(iteration, maybe_opt)| {
            if let Some(opt) = maybe_opt {
                Either::Left(
                    self.get_mapped_range(&opt, mapper, iteration as usize, snapshot)
                        .map(move |maybe_values| {
                            let next_opt = match &maybe_values {
                                Ok(values) => opt.next_mapped_range(values),
                                Err(..) => None,
                            };
                            Some((maybe_values, (iteration + 1, next_opt)))
                        }),
                )
            } else {
                Either::Right(future::ready(None))
            }
        })
    }

    /// Modify the database snapshot represented by transaction to remove all keys (if any) which
    /// are lexicographically greater than or equal to the given begin key and lexicographically
    /// less than the given end_key.
    ///
    /// The modification affects the actual database only if transaction is later committed with
    /// `Transaction::commit`.
    #[cfg_attr(
        feature = "trace",
        tracing::instrument(level = "debug", skip(self, begin, end))
    )]
    pub fn clear_range(&self, begin: &[u8], end: &[u8]) {
        unsafe {
            fdb_sys::fdb_transaction_clear_range(
                self.inner.as_ptr(),
                begin.as_ptr(),
                fdb_len(begin.len(), "begin"),
                end.as_ptr(),
                fdb_len(end.len(), "end"),
            )
        }

        if let Some(metrics) = &self.metrics {
            metrics.report_metrics(FdbCommand::ClearRange);
        }
    }

    /// Get the estimated byte size of the key range based on the byte sample collected by FDB
    #[cfg_api_versions(min = 630)]
    pub fn get_estimated_range_size_bytes(
        &self,
        begin: &[u8],
        end: &[u8],
    ) -> impl Future<Output = FdbResult<i64>> + Send + Sync + Unpin {
        FdbFuture::<i64>::new(unsafe {
            fdb_sys::fdb_transaction_get_estimated_range_size_bytes(
                self.inner.as_ptr(),
                begin.as_ptr(),
                fdb_len(begin.len(), "begin"),
                end.as_ptr(),
                fdb_len(end.len(), "end"),
            )
        })
    }

    /// Attempts to commit the sets and clears previously applied to the database snapshot
    /// represented by transaction to the actual database.
    ///
    /// The commit may or may not succeed – in particular, if a conflicting transaction previously
    /// committed, then the commit must fail in order to preserve transactional isolation. If the
    /// commit does succeed, the transaction is durably committed to the database and all
    /// subsequently started transactions will observe its effects.
    ///
    /// It is not necessary to commit a read-only transaction – you can simply drop it.
    ///
    /// Callers will usually want to retry a transaction if the commit or a another method on the
    /// transaction returns a retryable error (see `on_error` and/or `Database::transact`).
    ///
    /// As with other client/server databases, in some failure scenarios a client may be unable to
    /// determine whether a transaction succeeded. In these cases, `Transaction::commit` will return
    /// an error and `is_maybe_committed()` will returns true on that error. The `on_error` function
    /// treats this error as retryable, so retry loops that don’t check for `is_maybe_committed()`
    /// could execute the transaction twice. In these cases, you must consider the idempotence of
    /// the transaction. For more information, see [Transactions with unknown results](https://apple.github.io/foundationdb/developer-guide.html#developer-guide-unknown-results).
    ///
    /// Normally, commit will wait for outstanding reads to return. However, if those reads were
    /// snapshot reads or the transaction option for disabling “read-your-writes” has been invoked,
    /// any outstanding reads will immediately return errors.
    #[cfg_attr(feature = "trace", tracing::instrument(level = "debug", skip(self)))]
    pub fn commit(self) -> impl Future<Output = TransactionResult> + Send + Sync + Unpin {
        FdbFuture::<()>::new(unsafe { fdb_sys::fdb_transaction_commit(self.inner.as_ptr()) }).map(
            move |r| match r {
                Ok(()) => Ok(TransactionCommitted { tr: self }),
                Err(err) => Err(TransactionCommitError { tr: self, err }),
            },
        )
    }

    /// Implements the recommended retry and backoff behavior for a transaction. This function knows
    /// which of the error codes generated by other `Transaction` functions represent temporary
    /// error conditions and which represent application errors that should be handled by the
    /// application. It also implements an exponential backoff strategy to avoid swamping the
    /// database cluster with excessive retries when there is a high level of conflict between
    /// transactions.
    ///
    /// It is not necessary to call `reset()` when handling an error with `on_error()` since the
    /// transaction has already been reset.
    ///
    /// You should not call this method most of the times and use `Database::transact` which
    /// implements a retry loop strategy for you.
    pub fn on_error(
        self,
        err: FdbError,
    ) -> impl Future<Output = FdbResult<Transaction>> + Send + Sync + Unpin {
        FdbFuture::<()>::new(unsafe {
            fdb_sys::fdb_transaction_on_error(self.inner.as_ptr(), err.code())
        })
        .map_ok(|()| self)
    }

    /// Cancels the transaction. All pending or future uses of the transaction will return a
    /// transaction_cancelled error. The transaction can be used again after it is reset.
    pub fn cancel(self) -> TransactionCancelled {
        unsafe {
            fdb_sys::fdb_transaction_cancel(self.inner.as_ptr());
        }
        TransactionCancelled { tr: self }
    }

    /// Sets a custom metric for the transaction with the specified name, value, and labels.
    ///
    /// Custom metrics allow you to track application-specific metrics during transaction execution.
    /// These metrics are collected alongside the standard FoundationDB metrics and can be used
    /// for monitoring, debugging, and performance analysis.
    ///
    /// # Arguments
    /// * `name` - The name of the metric (e.g., "query_time", "cache_hits")
    /// * `value` - The value to set for the metric
    /// * `labels` - Key-value pairs for labeling the metric, allowing for dimensional metrics
    ///   (e.g., `[("operation", "read"), ("region", "us-west")]`)
    ///
    /// # Returns
    /// * `Ok(())` if the metric was set successfully
    /// * `Err(TransactionMetricsNotFound)` if this transaction was not created with metrics instrumentation
    ///
    /// # Example
    /// ```
    /// # use foundationdb::*;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let db = Database::default()?;
    /// let metrics = TransactionMetrics::new();
    /// let txn = db.create_instrumented_trx(metrics)?;
    ///
    /// // Set a custom metric with labels
    /// txn.set_custom_metric("query_processing_time", 42, &[("query_type", "select"), ("table", "users")])?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Note
    /// This method only works on transactions created with `create_instrumented_trx()` or within
    /// `instrumented_run()`. For regular transactions created with `create_trx()`, this will
    /// return `Err(TransactionMetricsNotFound)`.
    pub fn set_custom_metric(
        &self,
        name: &str,
        value: u64,
        labels: &[(&str, &str)],
    ) -> Result<(), TransactionMetricsNotFound> {
        if let Some(metrics) = &self.metrics {
            metrics.set_custom(name, value, labels);
            Ok(())
        } else {
            Err(TransactionMetricsNotFound)
        }
    }

    /// Increments a custom metric for the transaction by the specified amount.
    ///
    /// This is useful for counting events or accumulating values during transaction execution.
    /// If the metric doesn't exist yet, it will be created with the specified amount.
    /// If it already exists with the same labels, its value will be incremented.
    ///
    /// # Arguments
    /// * `name` - The name of the metric to increment (e.g., "requests", "bytes_processed")
    /// * `amount` - The amount to increment the metric by
    /// * `labels` - Key-value pairs for labeling the metric, allowing for dimensional metrics
    ///   (e.g., `[("status", "success"), ("endpoint", "api/v1/users")]`)
    ///
    /// # Returns
    /// * `Ok(())` if the metric was incremented successfully
    /// * `Err(TransactionMetricsNotFound)` if this transaction was not created with metrics instrumentation
    ///
    /// # Example
    /// ```
    /// # use foundationdb::*;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let db = Database::default()?;
    /// let metrics = TransactionMetrics::new();
    /// let txn = db.create_instrumented_trx(metrics)?;
    ///
    /// // Increment a counter each time a specific operation occurs
    /// txn.increment_custom_metric("cache_misses", 1, &[("cache", "user_data")])?;
    ///
    /// // Later in the transaction, increment it again
    /// txn.increment_custom_metric("cache_misses", 1, &[("cache", "user_data")])?;
    ///
    /// // The final value will be 2
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Note
    /// This method only works on transactions created with `create_instrumented_trx()` or within
    /// `instrumented_run()`. For regular transactions created with `create_trx()`, this will
    /// return `Err(TransactionMetricsNotFound)`.
    pub fn increment_custom_metric(
        &self,
        name: &str,
        amount: u64,
        labels: &[(&str, &str)],
    ) -> Result<(), TransactionMetricsNotFound> {
        if let Some(metrics) = &self.metrics {
            metrics.increment_custom(name, amount, labels);
            Ok(())
        } else {
            Err(TransactionMetricsNotFound)
        }
    }

    /// Returns a list of public network addresses as strings, one for each of the storage servers
    /// responsible for storing key_name and its associated value.
    pub fn get_addresses_for_key(
        &self,
        key: &[u8],
    ) -> impl Future<Output = FdbResult<FdbAddresses>> + Send + Sync + Unpin {
        FdbFuture::new(unsafe {
            fdb_sys::fdb_transaction_get_addresses_for_key(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
            )
        })
    }

    /// A watch's behavior is relative to the transaction that created it. A watch will report a
    /// change in relation to the key’s value as readable by that transaction. The initial value
    /// used for comparison is either that of the transaction’s read version or the value as
    /// modified by the transaction itself prior to the creation of the watch. If the value changes
    /// and then changes back to its initial value, the watch might not report the change.
    ///
    /// Until the transaction that created it has been committed, a watch will not report changes
    /// made by other transactions. In contrast, a watch will immediately report changes made by
    /// the transaction itself. Watches cannot be created if the transaction has set the
    /// READ_YOUR_WRITES_DISABLE transaction option, and an attempt to do so will return an
    /// watches_disabled error.
    ///
    /// If the transaction used to create a watch encounters an error during commit, then the watch
    /// will be set with that error. A transaction whose commit result is unknown will set all of
    /// its watches with the commit_unknown_result error. If an uncommitted transaction is reset or
    /// destroyed, then any watches it created will be set with the transaction_cancelled error.
    ///
    /// Returns an future representing an empty value that will be set once the watch has
    /// detected a change to the value at the specified key.
    ///
    /// By default, each database connection can have no more than 10,000 watches that have not yet
    /// reported a change. When this number is exceeded, an attempt to create a watch will return a
    /// too_many_watches error. This limit can be changed using the MAX_WATCHES database option.
    /// Because a watch outlives the transaction that creates it, any watch that is no longer
    /// needed should be cancelled by dropping its future.
    pub fn watch(&self, key: &[u8]) -> impl Future<Output = FdbResult<()>> + Send + Sync + Unpin {
        FdbFuture::new(unsafe {
            fdb_sys::fdb_transaction_watch(
                self.inner.as_ptr(),
                key.as_ptr(),
                fdb_len(key.len(), "key"),
            )
        })
    }

    /// Returns an FDBFuture which will be set to the approximate transaction size so far in the
    /// returned future, which is the summation of the estimated size of mutations, read conflict
    /// ranges, and write conflict ranges.
    ///
    /// This can be called multiple times before the transaction is committed.
    #[cfg_api_versions(min = 620)]
    pub fn get_approximate_size(
        &self,
    ) -> impl Future<Output = FdbResult<i64>> + Send + Sync + Unpin {
        FdbFuture::new(unsafe {
            fdb_sys::fdb_transaction_get_approximate_size(self.inner.as_ptr())
        })
    }

    /// Gets a list of keys that can split the given range into (roughly) equally sized chunks based on chunk_size.
    /// Note: the returned split points contain the start key and end key of the given range.
    #[cfg_api_versions(min = 700)]
    pub fn get_range_split_points(
        &self,
        begin: &[u8],
        end: &[u8],
        chunk_size: i64,
    ) -> impl Future<Output = FdbResult<FdbKeys>> + Send + Sync + Unpin {
        FdbFuture::<FdbKeys>::new(unsafe {
            fdb_sys::fdb_transaction_get_range_split_points(
                self.inner.as_ptr(),
                begin.as_ptr(),
                fdb_len(begin.len(), "begin"),
                end.as_ptr(),
                fdb_len(end.len(), "end"),
                chunk_size,
            )
        })
    }

    /// Returns an FDBFuture which will be set to the versionstamp which was used by any
    /// versionstamp operations in this transaction.
    ///
    /// The future will be ready only after the successful completion of a call to `commit()` on
    /// this Transaction. Read-only transactions do not modify the database when committed and will
    /// result in the future completing with an error. Keep in mind that a transaction which reads
    /// keys and then sets them to their current values may be optimized to a read-only transaction.
    ///
    /// Most applications will not call this function.
    pub fn get_versionstamp(
        &self,
    ) -> impl Future<Output = FdbResult<FdbSlice>> + Send + Sync + Unpin {
        FdbFuture::new(unsafe { fdb_sys::fdb_transaction_get_versionstamp(self.inner.as_ptr()) })
    }

    /// The transaction obtains a snapshot read version automatically at the time of the first call
    /// to `get_*()` (including this one) and (unless causal consistency has been deliberately
    /// compromised by transaction options) is guaranteed to represent all transactions which were
    /// reported committed before that call.
    pub fn get_read_version(&self) -> impl Future<Output = FdbResult<i64>> + Send + Sync + Unpin {
        FdbFuture::new(unsafe { fdb_sys::fdb_transaction_get_read_version(self.inner.as_ptr()) })
    }

    /// Sets the snapshot read version used by a transaction.
    ///
    /// This is not needed in simple cases.
    /// If the given version is too old, subsequent reads will fail with error_code_past_version;
    /// if it is too new, subsequent reads may be delayed indefinitely and/or fail with
    /// error_code_future_version. If any of get_*() have been called on this transaction already,
    /// the result is undefined.
    pub fn set_read_version(&self, version: i64) {
        unsafe { fdb_sys::fdb_transaction_set_read_version(self.inner.as_ptr(), version) }
    }

    /// The metadata version key `\xff/metadataVersion` is a key intended to help layers deal with hot keys.
    /// The value of this key is sent to clients along with the read version from the proxy,
    /// so a client can read its value without communicating with a storage server.
    /// To retrieve the metadataVersion, you need to set `TransactionOption::ReadSystemKeys`
    #[cfg_api_versions(min = 610)]
    pub async fn get_metadata_version(&self, snapshot: bool) -> FdbResult<Option<i64>> {
        match self.get(METADATA_VERSION_KEY, snapshot).await {
            Ok(Some(fdb_slice)) => {
                let value = fdb_slice.deref();
                // as we cannot write the metadata-key directly(we must mutate with an atomic_op),
                // can we assume that it will always be the correct size?
                if value.len() < 8 {
                    return Ok(None);
                }

                // The 80-bits versionstamps are 10 bytes longs, and are composed of:
                // * 8 bytes (Transaction Version)
                // * followed by 2 bytes (Transaction Batch Order)
                // More details can be found here: https://forums.foundationdb.org/t/implementing-versionstamps-in-bindings/250
                let mut arr = [0u8; 8];
                arr.copy_from_slice(&value[0..8]);
                let transaction_version: i64 = i64::from_be_bytes(arr);

                Ok(Some(transaction_version))
            }
            Ok(None) => Ok(None),

            Err(err) => Err(err),
        }
    }

    #[cfg_api_versions(min = 610)]
    pub fn update_metadata_version(&self) {
        // The param is transformed by removing the final four bytes from ``param`` and reading
        // those as a little-Endian 32-bit integer to get a position ``pos``.
        // The 10 bytes of the parameter from ``pos`` to ``pos + 10`` are replaced with the
        // versionstamp of the transaction used. The first byte of the parameter is position 0.
        // As we only have the metadata value, we can just create an 14-bytes Vec filled with 0u8.
        let param = vec![0u8; 14];
        self.atomic_op(
            METADATA_VERSION_KEY,
            param.as_slice(),
            options::MutationType::SetVersionstampedValue,
        )
    }

    /// Reset transaction to its initial state.
    ///
    /// In order to protect against a race condition with cancel(), this call require a mutable
    /// access to the transaction.
    ///
    /// This is similar to dropping the transaction and creating a new one.
    ///
    /// It is not necessary to call `reset()` when handling an error with `on_error()` since the
    /// transaction has already been reset.
    pub fn reset(&mut self) {
        unsafe { fdb_sys::fdb_transaction_reset(self.inner.as_ptr()) }
    }

    /// Adds a conflict range to a transaction without performing the associated read or write.
    ///
    /// # Note
    ///
    /// Most applications will use the serializable isolation that transactions provide by default
    /// and will not need to manipulate conflict ranges.
    pub fn add_conflict_range(
        &self,
        begin: &[u8],
        end: &[u8],
        ty: options::ConflictRangeType,
    ) -> FdbResult<()> {
        error::eval(unsafe {
            fdb_sys::fdb_transaction_add_conflict_range(
                self.inner.as_ptr(),
                begin.as_ptr(),
                fdb_len(begin.len(), "begin"),
                end.as_ptr(),
                fdb_len(end.len(), "end"),
                ty.code(),
            )
        })
    }
}

impl Drop for Transaction {
    fn drop(&mut self) {
        unsafe {
            fdb_sys::fdb_transaction_destroy(self.inner.as_ptr());
        }
    }
}

/// A retryable transaction, generated by Database.run
#[derive(Clone)]
pub struct RetryableTransaction {
    inner: Arc<Transaction>,
}

impl Deref for RetryableTransaction {
    type Target = Transaction;
    fn deref(&self) -> &Transaction {
        self.inner.deref()
    }
}

impl RetryableTransaction {
    pub(crate) fn new(t: Transaction) -> RetryableTransaction {
        RetryableTransaction { inner: Arc::new(t) }
    }

    pub(crate) fn take(self) -> Result<Transaction, FdbBindingError> {
        // checking weak references
        if Arc::weak_count(&self.inner) != 0 {
            return Err(FdbBindingError::ReferenceToTransactionKept);
        }
        Arc::try_unwrap(self.inner).map_err(|_| FdbBindingError::ReferenceToTransactionKept)
    }

    pub(crate) async fn on_error(
        self,
        err: FdbError,
    ) -> Result<Result<RetryableTransaction, FdbError>, FdbBindingError> {
        Ok(self
            .take()?
            .on_error(err)
            .await
            .map(RetryableTransaction::new))
    }

    #[cfg_attr(feature = "trace", tracing::instrument(level = "debug", skip(self)))]
    pub(crate) async fn commit(
        self,
    ) -> Result<Result<TransactionCommitted, TransactionCommitError>, FdbBindingError> {
        Ok(self.take()?.commit().await)
    }
}