lix 0.18.0

//! Engine session boundary.
//!
//! Transaction invariant: a session has one execution lease. Parent-handle
//! calls use it for implicit single-statement execution; explicit transactions
//! hold it until commit or rollback. Session feature submodules should enter
//! write flows through the centralized session helpers rather than opening
//! `Transaction` directly or using session-level read helpers inside writes.
//!
//! MVP boundary: session close can cancel queued or pre-boundary writes until
//! the storage commit point-of-no-return. After that point, close waits for
//! commit completion. Persistent writes await the backend durability boundary
//! by default; repository handles can explicitly select buffered acknowledgement.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

#[cfg(test)]
mod catalog_visibility_tests;
mod checkpoint;
mod context;
mod create_branch;
mod execute;
mod gc;
pub(crate) mod idempotency;
pub(crate) mod media_upload;
mod merge;
pub(crate) mod observe;
mod switch_branch;
mod transaction;

pub(crate) use media_upload::stage_reclaimable_upload_receipts;
#[cfg(feature = "storage-benches")]
pub(crate) use merge::{MergeCommitsForBench, analyze_merge_for_bench};
// Owner facade for the storage-space registry (`crate::storage_spaces`),
// which is compiled in every configuration.
pub(crate) use media_upload::{UPLOAD_MANIFEST_LEAF_SPACE, UPLOAD_STATE_SPACE};

pub(crate) use crate::common::ExecuteStatementMetadata;
#[cfg(feature = "server-protocol")]
pub(crate) use crate::common::VerifiedRequestBlob;
pub(crate) use checkpoint::CreateCheckpointReceipt;
pub use context::SessionContext;
pub(crate) use context::{SessionBranch, load_default_branch_id_from_index};
pub use create_branch::{CreateBranchOptions, CreateBranchReceipt};
pub use execute::{
    CoherentReadBatch, CommitReceipt, CommitSpan, ExecuteBatchResult, ExecuteBatchStatement,
    ExecuteOptions, ExecuteResult, ResultRowRef, Row, TryFromValue,
};
pub(crate) use execute::{ExecutionDisposition, FileRead};
pub(crate) use idempotency::ExecuteIdempotency;
pub(crate) use idempotency::{
    EXECUTE_IDEMPOTENCY_RECEIPT_SPACE, ExecuteIdempotencyReceipt, encode_receipt,
};
pub(crate) use media_upload::FileUploadProgress;
pub use merge::{
    MergeBranchOptions, MergeBranchOutcome, MergeBranchPreview, MergeBranchPreviewOptions,
    MergeBranchReceipt, MergeChangeStats,
};
pub use observe::ObserveEvent;
pub(crate) use observe::ObserveEvents as SessionObserveEvents;
pub use switch_branch::{SwitchBranchOptions, SwitchBranchReceipt};
pub use transaction::SessionTransaction;

/// Zero-cost adapter for futures that rustc cannot prove `Send` because an
/// opaque async call contains higher-ranked references. Construction is unsafe:
/// callers must verify that every value retained across suspension is `Send`.
#[repr(transparent)]
pub(crate) struct AssumeSendFuture<F>(F);

impl<F> AssumeSendFuture<F> {
    pub(crate) unsafe fn new(future: F) -> Self {
        Self(future)
    }
}

// SAFETY: `AssumeSendFuture::new` is private and unsafe; each call site must
// establish the wrapped future's complete suspension state is movable. Every
// current call site is pinned by a compile-time proof; see
// `session::execute::assume_send_future_proofs`.
//
// `F::Output: Send` is not part of that obligation, but is required here so the
// wrapper can never launder a non-`Send` *result* onto another thread.
unsafe impl<F> Send for AssumeSendFuture<F>
where
    F: Future,
    F::Output: Send,
{
}

impl<F> Future for AssumeSendFuture<F>
where
    F: Future,
{
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        // SAFETY: projecting through the transparent wrapper does not move F.
        unsafe { self.map_unchecked_mut(|wrapped| &mut wrapped.0) }.poll(context)
    }
}

/// A storage adapter whose `Read<'a>` genuinely borrows `'a`.
///
/// `Memory::Read<'a> = MemoryRead` is lifetime-independent, so proofs written
/// against `Memory` alone exercise the easy case: the higher-ranked obstruction
/// that forces `AssumeSendFuture` collapses before rustc ever gets there. The
/// shipping RocksDB adapter is `Read<'a> = RocksDBRead<'a>`, and
/// `FilesystemStorage` borrows in both `Read` and `Write`. This adapter
/// reproduces that shape over `Memory`'s storage so the `Send` proofs cover the
/// configuration that actually ships.
#[cfg(test)]
pub(crate) mod borrowing_proof_storage {
    use std::future::Future;

    use crate::storage_adapter::{
        Memory, MemoryRead, MemoryWrite, Storage, StorageBeginScanOptions, StorageError,
        StorageGetManyRequest, StorageGetManyResult, StorageKeyRange, StorageRead,
        StorageReadOptions, StorageScanCursor, StorageSpace, StorageWriteOptions,
    };

    /// Read handle carrying a real `'a` borrow of the owning storage.
    pub(crate) struct BorrowingRead<'a> {
        inner: MemoryRead,
        _borrow: &'a Memory,
    }

    impl StorageRead for BorrowingRead<'_> {
        fn snapshot_cache_key(&self) -> Option<u128> {
            self.inner.snapshot_cache_key()
        }

        fn get_many(
            &self,
            requests: &[StorageGetManyRequest<'_>],
        ) -> impl Future<Output = Result<StorageGetManyResult, StorageError>> + Send {
            self.inner.get_many(requests)
        }

        fn begin_scan(
            &self,
            space: StorageSpace,
            range: StorageKeyRange,
            opts: StorageBeginScanOptions,
        ) -> impl Future<Output = Result<StorageScanCursor<'_>, StorageError>> + Send {
            self.inner.begin_scan(space, range, opts)
        }
    }

    #[derive(Clone, Default)]
    pub(crate) struct BorrowingStorage(Memory);

    impl Storage for BorrowingStorage {
        type Read<'a>
            = BorrowingRead<'a>
        where
            Self: 'a;

        type Write<'a>
            = MemoryWrite
        where
            Self: 'a;

        fn acquire_session(
            &self,
        ) -> impl Future<Output = Result<crate::storage::StorageSessionToken, StorageError>> + Send
        {
            self.0.acquire_session()
        }

        fn begin_read(
            &self,
            opts: StorageReadOptions,
        ) -> impl Future<Output = Result<Self::Read<'_>, StorageError>> + Send {
            async move {
                Ok(BorrowingRead {
                    inner: self.0.begin_read(opts).await?,
                    _borrow: &self.0,
                })
            }
        }

        fn begin_write(
            &self,
            opts: StorageWriteOptions,
        ) -> impl Future<Output = Result<Self::Write<'_>, StorageError>> + Send {
            self.0.begin_write(opts)
        }
    }
}

pub(crate) use media_upload::{export_recoverable_uploads, has_recoverable_uploads};

pub(crate) use execute::{
    discover_read_fulfillment, prepare_partial_candidate_read_scope,
    seed_foreground_filesystem_interest,
};

pub(crate) use merge::{
    MergeAnalysis, analyze_incoming_rows, stage_merge_native_heads, stage_native_change_application,
};