veilid_core/veilid_api/

dht_transaction.rs

use super::*;
use crate::storage_manager::OutboundTransactionHandle;

impl_veilid_log_facility!("veilid_api");

///////////////////////////////////////////////////////////////////////////////////////

/// DHT Transactions the way you perform multiple simulateous atomic operations over a set of DHT records.
///
/// DHT operations performed out of a transaction may be processed in any order, and only operate on one subkey at a time
/// for a given record. Transactions allow you to bind a set of operations so they all succeed, or fail together, and at the same time.
///
/// Transactional DHT operations can only be performed when the node is online, and will error with [VeilidAPIError::TryAgain] if offline.
///
/// Transactions must be committed when all of their operations are registered, or rolled back if the group of operations is to be cancelled.
///
/// Each transaction holds a network-side resource that the caller must release by calling [DHTTransaction::commit] or [DHTTransaction::rollback]. Dropping a [DHTTransaction] without doing either logs a warning and tears the transaction down in the background.
#[derive(Clone)]
#[must_use]
pub struct DHTTransaction {
    /// API in use
    api: VeilidAPI,
    /// Inner transaction
    inner: Arc<Mutex<DHTTransactionInner>>,
}

impl fmt::Debug for DHTTransaction {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DHTTransaction")
            .field("handle", &self.inner.lock().opt_transaction_handle)
            .finish()
    }
}

impl DHTTransaction {
    ////////////////////////////////////////////////////////////////

    pub(super) fn new(api: VeilidAPI, handle: OutboundTransactionHandle) -> VeilidAPIResult<Self> {
        let registry = api.core_context()?.registry();
        Ok(Self {
            api,
            inner: Arc::new(Mutex::new(DHTTransactionInner {
                registry,
                opt_transaction_handle: Some(handle),
            })),
        })
    }

    /// Get the [VeilidAPI] object that created this [DHTTransaction].
    pub fn api(&self) -> VeilidAPI {
        self.api.clone()
    }

    #[must_use]
    pub(crate) fn log_key(&self) -> &str {
        self.api.log_key()
    }

    /// Extend the transaction with additional record keys
    ///
    /// Blocks on a network begin fanout for the added records and requires the node to be online. Idempotent for keys already in the transaction: returns `Ok(())` without network activity if no new records would be added.
    ///
    /// Errors with [VeilidAPIError::TransactionNotFound] if the transaction handle is already completed or unknown, [VeilidAPIError::MissingArgument] if `record_keys` contains duplicates, [VeilidAPIError::InvalidArgument] if the merged record set would exceed the per-transaction record limit, and [VeilidAPIError::TryAgain] (retry) if the node is offline or the begin fanout for the added records could not reach consensus.
    #[cfg_attr(feature = "instrument", instrument(target = "veilid_api", level = "debug", fields(duration, __VEILID_LOG_KEY = self.log_key(), transaction_handle), skip(self), ret))]
    pub async fn extend(
        &self,
        record_keys: Vec<RecordKey>,
        options: Option<TransactDHTRecordsOptions>,
    ) -> VeilidAPIResult<()> {
        async move {
            let storage_manager = self.api.core_context()?.storage_manager();
            let transaction_handle = {
                let inner = self.inner.lock();
                inner.opt_transaction_handle.ok_or_else(|| VeilidAPIError::transaction_not_found("transaction already completed"))?
            };
            tracing::Span::current().record("transaction_handle", transaction_handle.to_string());

            let recorder = DurationRecorder::new("DHTTransaction::extend", |name, start| {
                veilid_log!(self debug
                    "{}[start={:#}](transaction_handle: {}, record_keys: {:?}, options: {:?})", name, start, transaction_handle, record_keys, options);
            });
            recorder.record_fut(
                storage_manager.extend_transaction(transaction_handle, record_keys, options),
                |name, start, dur, ret| {
                    veilid_log!(self debug
                        "{}[start={:#} dur={:#}](ret: {:?})", name, start, dur, ret);
                    ret
                },
            ).await
        }.await.inspect_err(log_veilid_api_error!(self))
    }

    /// Commit the transaction
    /// All write operations are performed atomically
    ///
    /// Consumes the transaction and releases its network-side resource (the other half is [DHTTransaction::rollback]). Blocks on the end and commit consensus barriers and requires the node to be online. Completes the transaction exactly once: a second commit or a rollback errors with `transaction_not_found`.
    ///
    /// Errors with [VeilidAPIError::TransactionNotFound] if the transaction was already committed, rolled back, or is unknown, and [VeilidAPIError::TryAgain] (retry) if the node is offline or the end/commit barriers could not reach consensus.
    #[cfg_attr(feature = "instrument", instrument(target = "veilid_api", level = "debug", fields(duration, __VEILID_LOG_KEY = self.log_key(), transaction_handle), skip(self), ret))]
    pub async fn commit(self) -> VeilidAPIResult<()> {
        async {
            let storage_manager = self.api.core_context()?.storage_manager();
            let transaction_handle = {
                let mut inner = self.inner.lock();
                inner.opt_transaction_handle.take().ok_or_else(|| {
                    VeilidAPIError::transaction_not_found("transaction already completed")
                })?
            };
            tracing::Span::current().record("transaction_handle", transaction_handle.to_string());

            let recorder = DurationRecorder::new("DHTTransaction::commit", |name, start| {
                veilid_log!(self debug
                    "{}[start={:#}](transaction_handle: {})", name, start, transaction_handle);
            });
            recorder
                .record_fut(
                    Box::pin(storage_manager.end_and_commit_transaction(transaction_handle)),
                    |name, start, dur, ret| {
                        veilid_log!(self debug
                        "{}[start={:#} dur={:#}](ret: {:?})", name, start, dur, ret);
                        ret
                    },
                )
                .await
        }
        .await
        .inspect_err(log_veilid_api_error!(self))
    }

    /// Rollback the transaction
    /// No write operations are performed,
    ///
    /// Consumes the transaction and releases its network-side resource (the other half is [DHTTransaction::commit]). Blocks on sending rollbacks to the network and requires the node to be online. Completes the transaction exactly once: a second rollback or a commit errors with `transaction_not_found`.
    ///
    /// Errors with [VeilidAPIError::TransactionNotFound] if the transaction was already committed, rolled back, or is unknown, and [VeilidAPIError::TryAgain] (retry) if the node is offline.
    #[cfg_attr(feature = "instrument", instrument(target = "veilid_api", level = "debug", fields(duration, __VEILID_LOG_KEY = self.log_key(), transaction_handle), skip(self), ret))]
    pub async fn rollback(self) -> VeilidAPIResult<()> {
        async {
            let storage_manager = self.api.core_context()?.storage_manager();
            let transaction_handle = {
                let mut inner = self.inner.lock();
                inner.opt_transaction_handle.take().ok_or_else(|| {
                    VeilidAPIError::transaction_not_found("transaction already completed")
                })?
            };
            tracing::Span::current().record("transaction_handle", transaction_handle.to_string());

            let recorder = DurationRecorder::new("DHTTransaction::rollback", |name, start| {
                veilid_log!(self debug
                    "{}[start={:#}](transaction_handle: {})", name, start, transaction_handle);
            });
            recorder
                .record_fut(
                    Box::pin(storage_manager.rollback_transaction(transaction_handle)),
                    |name, start, dur, ret| {
                        veilid_log!(self debug
                        "{}[start={:#} dur={:#}](ret: {:?})", name, start, dur, ret);
                        ret
                    },
                )
                .await
        }
        .await
        .inspect_err(log_veilid_api_error!(self))
    }

    /// Add a set_dht_value operation to the transaction
    ///
    /// * Will fail if performed offline
    /// * Will fail if existing offline writes exist for this record key
    ///
    /// The writer, if specified, will override the 'default_writer' specified when the record is opened.
    ///
    /// Returns `None` if the value was successfully set.
    /// Returns `Some(data)` if the value set was older than the one available on the network.
    ///
    /// Blocks on the per-subkey lock (unbounded) and the set RPC to the transaction's node set, which retries non-responding nodes. Each per-node RPC is bounded by `network.rpc.timeout_ms`, but the lock wait and retry rounds are not, so the whole call has no single-timeout bound.
    ///
    /// Errors with [VeilidAPIError::TransactionNotFound] if the transaction handle is already completed or no longer in the Begin stage, [VeilidAPIError::InvalidArgument] if `record_key` is not open in the transaction or `subkey` is outside the schema range, [VeilidAPIError::Generic] if `record_key` is malformed (unsupported kind or bad length) or the subkey has no writer, and [VeilidAPIError::TryAgain] (retry) if the node is offline or write consensus was not reached this round. A non-responding node is retried rather than surfaced as a timeout.
    #[cfg_attr(feature = "instrument", instrument(target = "veilid_api", level = "debug", fields(duration, __VEILID_LOG_KEY = self.log_key(), transaction_handle, data.len = data.len()), skip(self, data), ret))]
    pub async fn set(
        &self,
        record_key: RecordKey,
        subkey: ValueSubkey,
        data: Vec<u8>,
        options: Option<DHTTransactionSetValueOptions>,
    ) -> VeilidAPIResult<Option<ValueData>> {
        async move {
            let storage_manager = self.api.core_context()?.storage_manager();
            let transaction_handle = {
                let inner = self.inner.lock();
                inner
                    .opt_transaction_handle
                    .ok_or_else(|| VeilidAPIError::transaction_not_found("transaction already completed"))?
            };
            tracing::Span::current().record("transaction_handle", transaction_handle.to_string());
            storage_manager.check_record_key(&record_key)?;

            let data_len = data.len();
            let recorder = DurationRecorder::new("DHTTransaction::set", |name, start| {
                veilid_log!(self debug
                    "{}[start={:#}](transaction_handle: {}, key: {}, subkey: {}, data: len={}, options: {:?})", name, start, transaction_handle, record_key, subkey, data_len, options);
            });
            recorder.record_fut(
                Box::pin(storage_manager.transaction_set(
                    transaction_handle,
                    record_key,
                    subkey,
                    data,
                    options,
                )),
                |name, start, dur, ret| {
                    veilid_log!(self debug
                        "{}[start={:#} dur={:#}](ret: {:?})", name, start, dur, ret);
                    ret
                },
            ).await
        }.await.inspect_err(log_veilid_api_error!(self))
    }

    /// Perform a get_dht_value operation inside the transaction
    ///
    /// * Will fail if performed offline
    /// * Will pull the latest value from the network, will fail if the local value is newer
    /// * Will fail if existing offline writes exist for this record key
    ///
    /// Returns `None` if the value subkey has not yet been set.
    /// Returns `Some(data)` if the value subkey has valid data.
    ///
    /// Blocks on the per-subkey lock (unbounded) and the get RPC to the transaction's node set, which retries non-responding nodes. Each per-node RPC is bounded by `network.rpc.timeout_ms`, but the lock wait and retry rounds are not, so the whole call has no single-timeout bound.
    ///
    /// Errors with [VeilidAPIError::TransactionNotFound] if the transaction handle is already completed or no longer in the Begin stage, [VeilidAPIError::InvalidArgument] if `record_key` is not in the transaction or `subkey` is outside the schema range, [VeilidAPIError::Generic] if `record_key` is malformed (unsupported kind or bad length), and [VeilidAPIError::TryAgain] (retry) if the node is offline or the network did not return the value that existed at begin time. A non-responding node is retried rather than surfaced as a timeout.
    #[cfg_attr(feature = "instrument", instrument(target = "veilid_api", level = "debug", fields(duration, __VEILID_LOG_KEY = self.log_key()), skip(self), ret))]
    pub async fn get(
        &self,
        record_key: RecordKey,
        subkey: ValueSubkey,
    ) -> VeilidAPIResult<Option<ValueData>> {
        async move {
            let storage_manager = self.api.core_context()?.storage_manager();
            let transaction_handle = {
                let inner = self.inner.lock();
                inner
                    .opt_transaction_handle
                    .ok_or_else(|| VeilidAPIError::transaction_not_found("transaction already completed"))?
            };
            tracing::Span::current().record("transaction_handle", transaction_handle.to_string());
            storage_manager.check_record_key(&record_key)?;

            let recorder = DurationRecorder::new("DHTTransaction::get", |name, start| {
                veilid_log!(self debug
                    "{}[start={:#}](transaction_handle: {}, key: {}, subkey: {})", name, start, transaction_handle, record_key, subkey);
            });
            recorder.record_fut(
                Box::pin(storage_manager.transaction_get(transaction_handle, record_key, subkey)),
                |name, start, dur, ret| {
                    veilid_log!(self debug
                        "{}[start={:#} dur={:#}](ret: {:?})", name, start, dur, ret);
                    ret
                },
            ).await
        }.await.inspect_err(log_veilid_api_error!(self))
    }

    /// Perform a inspect_dht_record operation inside the transaction
    ///
    /// * Does not perform any network activity, as the transaction state keeps all of the required information after the begin
    ///
    /// For information on arguments, see [RoutingContext::inspect_dht_record]
    ///
    /// Returns a DHTRecordReport with the subkey ranges that were returned that overlapped the schema, and sequence numbers for each of the subkeys in the range.
    ///
    /// Errors with [VeilidAPIError::TransactionNotFound] if the transaction handle is already completed, unknown, or no longer in the Begin stage (End, Commit, Rollback, or Failed), [VeilidAPIError::InvalidArgument] if `record_key` is not in the transaction, and [VeilidAPIError::Generic] if `record_key` is malformed (unsupported kind or bad length) or the transaction has not started. Performs no network activity and cannot time out.
    #[cfg_attr(feature = "instrument", instrument(target = "veilid_api", level = "debug", fields(duration, __VEILID_LOG_KEY = self.log_key(), transaction_handle), skip(self), ret))]
    pub async fn inspect(
        &self,
        record_key: RecordKey,
        subkeys: Option<ValueSubkeyRangeSet>,
        scope: DHTReportScope,
    ) -> VeilidAPIResult<DHTRecordReport> {
        async move {
            let storage_manager = self.api.core_context()?.storage_manager();
            let transaction_handle = {
                let inner = self.inner.lock();
                inner
                    .opt_transaction_handle
                    .ok_or_else(|| VeilidAPIError::transaction_not_found("transaction already completed"))?
            };
            tracing::Span::current().record("transaction_handle", transaction_handle.to_string());
            storage_manager.check_record_key(&record_key)?;

            let recorder = DurationRecorder::new("DHTTransaction::inspect", |name, start| {
                veilid_log!(self debug
                    "{}[start={:#}](transaction_handle: {}, record_key: {}, subkeys: {}, scope: {:?})", name, start, transaction_handle, record_key, subkeys.as_ref().map(|x| x.to_string()).unwrap_or_else(|| "None".to_string()), scope);
            });
            recorder.record(
                || storage_manager.transaction_inspect(transaction_handle, record_key, subkeys, scope),
                |name, start, dur, ret| {
                    veilid_log!(self debug
                        "{}[start={:#} dur={:#}](ret: {:?})", name, start, dur, ret);
                    ret
                },
            )
        }.await.inspect_err(log_veilid_api_error!(self))
    }
}
//////////////////////////////////////////////////////////////////////////////////////

struct DHTTransactionInner {
    registry: VeilidComponentRegistry,
    opt_transaction_handle: Option<OutboundTransactionHandle>,
}

impl Drop for DHTTransactionInner {
    fn drop(&mut self) {
        if let Some(transaction_handle) = self.opt_transaction_handle.take() {
            let registry = &self.registry;
            veilid_log!(registry warn "Dropped DHT transaction without commit or rollback");

            let storage_manager = registry.storage_manager();
            storage_manager.drop_transaction_sync(transaction_handle);
        }
    }
}