lix 0.18.0

use std::sync::Arc;

use smallvec::SmallVec;

use crate::catalog::CatalogFingerprint;
use crate::functions::{DeterministicRuntimeGuard, FunctionContext};
use crate::observe_invalidation::ObserveInvalidation;
use crate::sql2::SqlPlanningCache;
use crate::storage_adapter::Memory;
use crate::storage_adapter::Storage;
use crate::telemetry::{
    ActiveTelemetrySpan, Status, TRANSACTION_NOTIFY, TelemetryAttribute, TelemetrySink,
};
use tokio::sync::Notify;

use crate::LixError;
#[cfg(test)]
use crate::transaction::CommitBoundaryGuard;
use crate::transaction::{
    CommitBoundaryState, Transaction, TransactionCommitBoundary,
    open_transaction_with_runtime_boundary,
};
#[cfg(test)]
use crate::transaction_types::{RawWriteBatch, TransactionWriteRow};

use super::SessionContext;
use super::context::{SessionWriteAccess, closed_error};
use crate::sync::SyncModeState;
use crate::transaction::CommitCoordinator;

#[expect(missing_debug_implementations)]
pub struct SessionTransaction<StorageImpl: Storage + 'static = Memory> {
    pub(super) session: SessionContext<StorageImpl>,
    pub(super) transaction: Option<Transaction<StorageImpl>>,
    pub(super) runtime_functions: FunctionContext,
    transaction_manager: SessionTransactionManager,
    observe_invalidation: Arc<ObserveInvalidation>,
    pub(super) sql_planning_cache: Arc<SqlPlanningCache<CatalogFingerprint>>,
    _deterministic_runtime_guard: Option<DeterministicRuntimeGuard>,
    write_access: Option<SessionWriteAccess>,
    commit_coordinator: Arc<CommitCoordinator<StorageImpl>>,
    pub(super) telemetry: Option<Arc<dyn TelemetrySink>>,
    pub(super) sync_mode: SyncModeState,
    pub(super) sync_demand_tx: Option<tokio::sync::mpsc::Sender<crate::sync::SyncDemand>>,
    pub(super) has_started_statement: bool,
    pub(super) has_written_statement: bool,
    /// Reusable storage only for SQL literals containing doubled quote
    /// escapes. Ordinary warm literals continue to borrow the SQL text.
    pub(super) prepared_literal_escape_scratch: SmallVec<[String; 4]>,
    pub(super) prepared_literal_shape: crate::sql2::CachedUpdateLiteralShape,
}

impl<StorageImpl> SessionContext<StorageImpl>
where
    StorageImpl: Storage + Clone + Send + Sync + 'static,
{
    pub async fn begin_transaction(&self) -> Result<SessionTransaction<StorageImpl>, LixError> {
        self.ensure_open()?;
        let mut write_access = self.begin_explicit_session_write_access().await?;
        // Mode and sequence are global write authority. Serialize through the
        // same collaboration gate as every writer while the transaction's one
        // retained opening snapshot loads them. The runtime boundary below is
        // invoked immediately after that load: deterministic transactions keep
        // collaboration serialization and then take the runtime gate in the
        // global order, while ordinary transactions release serialization
        // before catalog and branch-head setup continues on the same snapshot.
        write_access
            .serialize_collaboration_writes(&self.collaboration_write_gate)
            .await;
        let (mut opened, deterministic_runtime_guard) =
            match open_transaction_with_runtime_boundary(
                &self.branch,
                self.active_account_id.to_string(),
                self.storage.clone(),
                Arc::clone(&self.hot_state),
                Arc::clone(&self.tracked_state),
                Arc::clone(&self.binary_cas),
                self.plugin_host.clone(),
                Arc::clone(&self.branch_ctx),
                Arc::clone(&self.catalog_context),
                Arc::clone(&self.sql_planning_cache),
                self.file_views.clone(),
                self.account_insertion.clone(),
                async |runtime_functions| {
                    if runtime_functions.deterministic_mode_enabled() {
                        Ok(Some(self.lock_deterministic_runtime().await))
                    } else {
                        write_access.release_collaboration_write_serialization();
                        Ok(None)
                    }
                },
            )
            .await
            {
                Ok(opened) => opened,
                Err(error) => {
                    return Err(error);
                }
            };
        self.ensure_open()?;
        let sync_role = self.sync_mode.role();
        let replica_remote_id = sync_role
            .is_replica()
            .then(|| self.sync_mode.replica_remote_id())
            .flatten();
        opened.transaction.set_sync_mode(
            sync_role,
            replica_remote_id,
            self.sync_mode.partial_admission(),
        );
        // An explicit SQL commit returns its own exact span; combining it
        // with another transaction would include unrelated changes.
        opened.transaction.require_individual_commit_span();
        opened
            .transaction
            .attach_commit_boundary(self.transaction_commit_boundary());
        self.transaction_manager()
            .mark_explicit_transaction_open()?;
        Ok(SessionTransaction {
            session: self.clone(),
            transaction: Some(opened.transaction),
            runtime_functions: opened.runtime_functions,
            transaction_manager: self.transaction_manager(),
            observe_invalidation: Arc::clone(&self.observe_invalidation),
            sql_planning_cache: Arc::clone(&self.sql_planning_cache),
            _deterministic_runtime_guard: deterministic_runtime_guard,
            write_access: Some(write_access),
            commit_coordinator: Arc::clone(&self.commit_coordinator),
            telemetry: self.telemetry.clone(),
            sync_mode: self.sync_mode.clone(),
            sync_demand_tx: None,
            has_started_statement: false,
            has_written_statement: false,
            prepared_literal_escape_scratch: SmallVec::new(),
            prepared_literal_shape: crate::sql2::CachedUpdateLiteralShape::default(),
        })
    }
}

impl<StorageImpl> SessionTransaction<StorageImpl>
where
    StorageImpl: Storage + Clone + Send + Sync + 'static,
{
    pub(crate) fn with_sync_demand_sender(
        mut self,
        sender: Option<tokio::sync::mpsc::Sender<crate::sync::SyncDemand>>,
    ) -> Self {
        if let Some(transaction) = self.transaction.as_mut() {
            transaction.set_sync_demand_sender(sender.clone());
        }
        self.sync_demand_tx = sender;
        self
    }

    pub(super) fn transaction_mut(&mut self) -> Result<&mut Transaction<StorageImpl>, LixError> {
        self.ensure_session_open()?;
        self.transaction
            .as_mut()
            .ok_or_else(|| transaction_state_error("Lix transaction is closed"))
    }

    #[cfg(test)]
    pub(crate) async fn stage_test_row(
        &mut self,
        row: TransactionWriteRow,
    ) -> Result<(), LixError> {
        self.transaction_mut()?
            .stage_engine_test_rows(RawWriteBatch::from_test_rows(vec![row]))
            .await?;
        Ok(())
    }

    #[cfg(test)]
    pub(crate) async fn restore_branch_ref_for_test(
        &mut self,
        branch_id: &str,
        expected_head_commit_id: &str,
        target_commit_id: &str,
    ) -> Result<(), LixError> {
        let expected_head_commit_id = crate::changelog::CommitId::parse_lix(
            expected_head_commit_id,
            "test restore expected branch head",
        )?;
        let target_commit_id = crate::changelog::CommitId::parse_lix(
            target_commit_id,
            "test restore target branch head",
        )?;
        if expected_head_commit_id == target_commit_id {
            return Ok(());
        }
        self.transaction_mut()?
            .restore_branch_ref(branch_id, expected_head_commit_id, target_commit_id)
            .await
    }

    pub fn active_branch_id(&self) -> Result<&str, LixError> {
        self.ensure_session_open()?;
        self.transaction
            .as_ref()
            .map(Transaction::active_branch_id)
            .ok_or_else(|| transaction_state_error("Lix transaction is closed"))
    }

    pub fn commit(self) -> impl Future<Output = Result<crate::CommitReceipt, LixError>> {
        // Callers can join independent commits. Return only the pinned owner,
        // so each caller does not retain another complete adapter-specific
        // preparation/publication state machine on its thread stack.
        Box::pin(self.commit_inner())
    }

    async fn commit_inner(mut self) -> Result<crate::CommitReceipt, LixError> {
        // Explicit transactions may remain open across application awaits, so
        // they do not hold the collaboration gate while planning. Serialize
        // their commit-time revalidation and persistence with bounded writes.
        // Deterministic transactions already retain this gate to preserve the
        // global lock order with their deterministic-runtime guard.
        let already_serialized = self
            .write_access
            .as_ref()
            .is_some_and(SessionWriteAccess::serializes_collaboration_writes);
        let operation_guard = self.begin_session_commit_operation()?;
        // The hydration path retains adapter-specific read futures. Keep it
        // outside the commit future, which callers may join for independent
        // transactions on the default thread stack.
        Box::pin(self.flush_prepared_mutations_with_sync()).await?;
        let transaction = self
            .transaction
            .take()
            .ok_or_else(|| transaction_state_error("Lix transaction is closed"))?;
        let result = if already_serialized {
            transaction.commit(&self.runtime_functions).await
        } else {
            self.commit_coordinator
                .commit(transaction, self.runtime_functions)
                .await
        };
        drop(operation_guard);
        let outcome = result?;
        drop(self.write_access.take());
        {
            let notify = already_serialized
                .then(|| {
                    ActiveTelemetrySpan::start_current(
                        &TRANSACTION_NOTIFY,
                        vec![
                            TelemetryAttribute::i64("lix.transaction.count", 1),
                            TelemetryAttribute::string(
                                "lix.commit_cohort_id",
                                outcome.commit_cohort_id.clone().unwrap_or_default(),
                            ),
                        ],
                    )
                })
                .flatten();
            let _entered = notify.as_ref().map(ActiveTelemetrySpan::enter);
            self.observe_invalidation
                .bump_if_storage_changed(&outcome.storage_stats);
            // Explicit transactions publish the same canonical event lane as
            // automatic writes. Wake server long-polls only after the durable
            // commit so readers never race a still-staged event.
            self.sync_mode.notify_sync_change();
            drop(_entered);
            if let Some(notify) = notify {
                notify.finish(Status::Unset, Vec::new());
            }
        }
        if let Some(checkpoint_sequence) = outcome.checkpoint_gc_sequence {
            self.session
                .schedule_checkpoint_gc_after_commit(checkpoint_sequence)
                .await;
        }
        let receipt = crate::CommitReceipt {
            commit: self
                .has_written_statement
                .then_some(outcome.active_branch_commit_span)
                .flatten()
                .map(crate::CommitSpan::from_commit_ids),
        };
        self.session
            .flush_partial_read_interests()
            .await
            .map_err(|error| receipt.annotate_completion_error(error))?;
        Ok(receipt)
    }

    pub async fn rollback(mut self) -> Result<(), LixError> {
        let transaction = self
            .transaction
            .take()
            .ok_or_else(|| transaction_state_error("Lix transaction is closed"))?;

        transaction.rollback().await?;
        self.session.flush_partial_read_interests().await
    }

    pub(super) fn ensure_session_open(&self) -> Result<(), LixError> {
        self.transaction_manager.ensure_open()
    }

    #[cfg(test)]
    pub(super) fn begin_session_operation(&self) -> Result<SessionOperationGuard, LixError> {
        self.transaction_manager.begin_transaction_operation()
    }

    fn begin_session_commit_operation(&self) -> Result<SessionOperationGuard, LixError> {
        self.transaction_manager
            .begin_transaction_commit_operation()
    }
}

pub(crate) fn transaction_state_error(message: impl Into<String>) -> LixError {
    LixError::new("LIX_INVALID_TRANSACTION_STATE", message)
}

#[derive(Clone)]
pub(super) struct SessionTransactionManager {
    inner: Arc<SessionTransactionManagerInner>,
}

struct SessionTransactionManagerInner {
    state: std::sync::Mutex<SessionTransactionState>,
    state_changed: Notify,
    commit_boundary: CommitBoundaryState,
}

#[derive(Debug, Default)]
enum SessionTransactionState {
    #[default]
    OpenIdle,
    OpenOperation {
        active_operations: usize,
    },
    OpenTransaction {
        active_operations: usize,
        owner: TransactionOwner,
    },
    Closing {
        active_operations: usize,
    },
    Closed,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum TransactionOwner {
    Automatic,
    ExplicitOpening,
    Explicit,
    ExplicitCommitting,
}

impl SessionTransactionManager {
    pub(super) fn new() -> Self {
        Self {
            inner: Arc::new(SessionTransactionManagerInner {
                state: std::sync::Mutex::new(SessionTransactionState::default()),
                state_changed: Notify::new(),
                commit_boundary: CommitBoundaryState::new(),
            }),
        }
    }

    pub(super) async fn close(&self) -> Result<(), LixError> {
        let mut commit_rx = self.inner.commit_boundary.subscribe();
        loop {
            let commit_gate = if self.inner.commit_boundary.is_active() {
                None
            } else {
                self.inner.commit_boundary.try_lock_commit()
            };
            if let Some(_commit_gate) = commit_gate {
                {
                    let mut state = self.lock_state();
                    if state.has_explicit_transaction() {
                        return Err(active_transaction_error());
                    }
                    let active_operations = state.active_operations();
                    *state = if active_operations == 0 {
                        SessionTransactionState::Closed
                    } else {
                        SessionTransactionState::Closing { active_operations }
                    };
                }
                self.inner.state_changed.notify_waiters();
                break;
            }

            let notified = self.inner.state_changed.notified();
            tokio::select! {
                () = notified => {}
                result = commit_rx.changed() => {
                    if result.is_err() {
                        self.inner.state_changed.notify_waiters();
                    }
                }
            }
        }

        loop {
            let notified = self.inner.state_changed.notified();
            {
                let mut state = self.lock_state();
                let commit_count = *commit_rx.borrow_and_update();
                if state.active_operations() == 0
                    && commit_count == 0
                    && !self.inner.commit_boundary.is_active()
                {
                    *state = SessionTransactionState::Closed;
                    break;
                }
            }
            tokio::select! {
                () = notified => {}
                result = commit_rx.changed() => {
                    if result.is_err() {
                        self.inner.state_changed.notify_waiters();
                    }
                }
            }
        }
        Ok(())
    }

    pub(super) fn is_closed(&self) -> bool {
        self.lock_state().is_closed()
    }

    pub(super) fn ensure_open(&self) -> Result<(), LixError> {
        if self.is_closed() {
            return Err(closed_error());
        }
        Ok(())
    }

    pub(super) fn ensure_observe_registration_allowed(&self) -> Result<(), LixError> {
        self.lock_state().ensure_observe_registration_allowed()
    }

    pub(super) async fn begin_waitable_session_operation(
        &self,
    ) -> Result<SessionOperationGuard, LixError> {
        loop {
            let notified = self.inner.state_changed.notified();
            let should_wait = {
                let mut state = self.lock_state();
                if state.is_automatic_transaction_in_progress() {
                    true
                } else {
                    state.begin_operation(SessionOperationScope::Session)?;
                    false
                }
            };
            if should_wait {
                notified.await;
                continue;
            }
            self.inner.state_changed.notify_waiters();

            if let Err(error) = self.ensure_open() {
                self.finish_operation();
                return Err(error);
            }

            return Ok(SessionOperationGuard {
                manager: self.clone(),
                read_interest_operation: None,
            });
        }
    }

    #[cfg(test)]
    pub(super) fn begin_transaction_operation(&self) -> Result<SessionOperationGuard, LixError> {
        self.begin_operation(SessionOperationScope::Transaction)
    }

    pub(super) fn begin_transaction_commit_operation(
        &self,
    ) -> Result<SessionOperationGuard, LixError> {
        self.begin_operation(SessionOperationScope::TransactionCommit)
    }

    fn begin_operation(
        &self,
        scope: SessionOperationScope,
    ) -> Result<SessionOperationGuard, LixError> {
        {
            let mut state = self.lock_state();
            state.begin_operation(scope)?;
        }
        self.inner.state_changed.notify_waiters();

        if let Err(error) = self.ensure_open() {
            self.finish_operation();
            return Err(error);
        }

        Ok(SessionOperationGuard {
            manager: self.clone(),
            read_interest_operation: None,
        })
    }

    pub(super) async fn begin_write_lease(&self) -> Result<SessionWriteLease, LixError> {
        loop {
            let notified = self.inner.state_changed.notified();
            let wait_for_session_operation = {
                let mut state = self.lock_state();
                if state.is_session_operation_in_progress()
                    || state.is_automatic_transaction_in_progress()
                {
                    true
                } else {
                    state.begin_write_lease(TransactionOwner::Automatic)?;
                    false
                }
            };
            if wait_for_session_operation {
                notified.await;
                continue;
            }
            self.inner.state_changed.notify_waiters();
            return self.open_reserved_write_lease();
        }
    }

    pub(super) fn begin_explicit_write_lease(&self) -> Result<SessionWriteLease, LixError> {
        self.begin_write_lease_for(TransactionOwner::ExplicitOpening)
    }

    fn begin_write_lease_for(
        &self,
        owner: TransactionOwner,
    ) -> Result<SessionWriteLease, LixError> {
        {
            let mut state = self.lock_state();
            state.begin_write_lease(owner)?;
        }
        self.inner.state_changed.notify_waiters();

        self.open_reserved_write_lease()
    }

    fn open_reserved_write_lease(&self) -> Result<SessionWriteLease, LixError> {
        let operation_guard = SessionOperationGuard {
            manager: self.clone(),
            read_interest_operation: None,
        };
        if let Err(error) = self.ensure_open() {
            drop(operation_guard);
            self.finish_transaction();
            return Err(error);
        }
        let transaction_guard = SessionTransactionGuard {
            manager: self.clone(),
        };
        Ok(SessionWriteLease {
            _transaction_guard: transaction_guard,
            _operation_guard: operation_guard,
        })
    }

    #[cfg(test)]
    pub(super) fn begin_commit(&self) -> CommitBoundaryGuard {
        self.inner.commit_boundary.begin()
    }

    pub(super) fn mark_explicit_transaction_open(&self) -> Result<(), LixError> {
        {
            let mut state = self.lock_state();
            state.mark_explicit_transaction_open()?;
        }
        self.inner.state_changed.notify_waiters();
        Ok(())
    }

    pub(super) fn transaction_commit_boundary(&self) -> TransactionCommitBoundary {
        let manager = self.clone();
        TransactionCommitBoundary::new(
            self.inner.commit_boundary.clone(),
            Arc::new(move || manager.ensure_open()),
        )
    }

    fn finish_operation(&self) {
        {
            let mut state = self.lock_state();
            state.finish_operation();
        }
        self.inner.state_changed.notify_waiters();
    }

    fn finish_transaction(&self) {
        {
            let mut state = self.lock_state();
            state.finish_transaction();
        }
        self.inner.state_changed.notify_waiters();
    }

    fn lock_state(&self) -> std::sync::MutexGuard<'_, SessionTransactionState> {
        self.inner
            .state
            .lock()
            .expect("session transaction manager lock should not poison")
    }

    #[cfg(test)]
    pub(super) fn operation_count_for_test(&self) -> usize {
        self.lock_state().active_operations()
    }

    #[cfg(test)]
    pub(super) fn commit_in_progress_for_test(&self) -> bool {
        self.inner.commit_boundary.is_active()
    }

    #[cfg(test)]
    pub(super) fn active_transaction_for_test(&self) -> bool {
        matches!(
            *self.lock_state(),
            SessionTransactionState::OpenTransaction { .. }
        )
    }
}

impl SessionTransactionState {
    fn is_closed(&self) -> bool {
        matches!(self, Self::Closing { .. } | Self::Closed)
    }

    fn active_operations(&self) -> usize {
        match self {
            Self::OpenIdle | Self::Closed => 0,
            Self::OpenOperation { active_operations }
            | Self::OpenTransaction {
                active_operations, ..
            }
            | Self::Closing { active_operations } => *active_operations,
        }
    }

    fn has_explicit_transaction(&self) -> bool {
        matches!(
            self,
            Self::OpenTransaction {
                owner: TransactionOwner::Explicit,
                ..
            }
        )
    }

    fn is_session_operation_in_progress(&self) -> bool {
        matches!(self, Self::OpenOperation { .. })
    }

    fn is_automatic_transaction_in_progress(&self) -> bool {
        matches!(
            self,
            Self::OpenTransaction {
                owner: TransactionOwner::Automatic,
                ..
            }
        )
    }

    fn ensure_observe_registration_allowed(&self) -> Result<(), LixError> {
        match self {
            Self::OpenIdle
            | Self::OpenOperation { .. }
            | Self::OpenTransaction {
                owner: TransactionOwner::Automatic,
                ..
            } => Ok(()),
            Self::OpenTransaction { .. } => Err(active_transaction_error()),
            Self::Closing { .. } | Self::Closed => Err(closed_error()),
        }
    }

    fn begin_operation(&mut self, scope: SessionOperationScope) -> Result<(), LixError> {
        match self {
            Self::OpenIdle => {
                if matches!(scope, SessionOperationScope::Session) {
                    *self = Self::OpenOperation {
                        active_operations: 1,
                    };
                    Ok(())
                } else {
                    Err(active_transaction_error())
                }
            }
            Self::OpenOperation { active_operations } => {
                if matches!(scope, SessionOperationScope::Session) {
                    *active_operations += 1;
                    Ok(())
                } else {
                    Err(active_transaction_error())
                }
            }
            Self::OpenTransaction {
                active_operations,
                owner,
            } => match scope {
                SessionOperationScope::Session => Err(active_transaction_error()),
                #[cfg(test)]
                SessionOperationScope::Transaction => {
                    *active_operations += 1;
                    Ok(())
                }
                SessionOperationScope::TransactionCommit => {
                    if *owner != TransactionOwner::Explicit {
                        return Err(active_transaction_error());
                    }
                    *owner = TransactionOwner::ExplicitCommitting;
                    *active_operations += 1;
                    Ok(())
                }
            },
            Self::Closing { .. } | Self::Closed => Err(closed_error()),
        }
    }

    fn begin_write_lease(&mut self, owner: TransactionOwner) -> Result<(), LixError> {
        match self {
            Self::OpenIdle => {
                *self = Self::OpenTransaction {
                    active_operations: 1,
                    owner,
                };
                Ok(())
            }
            Self::OpenOperation { .. } | Self::OpenTransaction { .. } => {
                Err(active_transaction_error())
            }
            Self::Closing { .. } | Self::Closed => Err(closed_error()),
        }
    }

    fn mark_explicit_transaction_open(&mut self) -> Result<(), LixError> {
        match self {
            Self::OpenTransaction {
                active_operations: 1,
                owner,
            } if *owner == TransactionOwner::ExplicitOpening => {
                *owner = TransactionOwner::Explicit;
                Ok(())
            }
            Self::Closing { .. } | Self::Closed => Err(closed_error()),
            _ => {
                panic!("explicit transaction should be opening before it is marked open");
            }
        }
    }

    fn finish_operation(&mut self) {
        match self {
            Self::OpenOperation { active_operations } => {
                *active_operations = active_operations
                    .checked_sub(1)
                    .expect("session operation count should not underflow");
                if *active_operations == 0 {
                    *self = Self::OpenIdle;
                }
            }
            Self::OpenTransaction {
                active_operations, ..
            } => {
                *active_operations = active_operations
                    .checked_sub(1)
                    .expect("session operation count should not underflow");
            }
            Self::Closing { active_operations } => {
                *active_operations = active_operations
                    .checked_sub(1)
                    .expect("session operation count should not underflow");
                if *active_operations == 0 {
                    *self = Self::Closed;
                }
            }
            Self::OpenIdle | Self::Closed => {
                panic!("session operation count should not underflow");
            }
        }
    }

    fn finish_transaction(&mut self) {
        match self {
            Self::OpenTransaction {
                active_operations: 0,
                ..
            } => {
                *self = Self::OpenIdle;
            }
            Self::OpenTransaction { .. } | Self::Closing { .. } | Self::Closed => {}
            Self::OpenIdle | Self::OpenOperation { .. } => {
                panic!("session transaction should be active before it is finished");
            }
        }
    }
}

#[derive(Clone, Copy)]
enum SessionOperationScope {
    Session,
    #[cfg(test)]
    Transaction,
    TransactionCommit,
}

fn active_transaction_error() -> LixError {
    transaction_state_error(
        "Lix handle has an active transaction; use the transaction handle for reads and writes until it is committed or rolled back",
    )
}

pub(super) struct SessionWriteLease {
    _operation_guard: SessionOperationGuard,
    _transaction_guard: SessionTransactionGuard,
}

pub(super) struct SessionTransactionGuard {
    manager: SessionTransactionManager,
}

impl Drop for SessionTransactionGuard {
    fn drop(&mut self) {
        self.manager.finish_transaction();
    }
}

pub(crate) struct SessionOperationGuard {
    manager: SessionTransactionManager,
    pub(super) read_interest_operation: Option<crate::hot_state::ReadInterestOperation>,
}

impl Drop for SessionOperationGuard {
    fn drop(&mut self) {
        self.manager.finish_operation();
    }
}