saddle-core 0.3.8

//! Opaque cross-component authority for Database physical disposition.
//!
//! Core owns only the linear pairing facts. Database owns physical I/O and
//! Runtime owns the admitted request; neither component can recreate the
//! other's half from identifiers or summaries.

use std::sync::{
    Arc,
    atomic::{AtomicU64, Ordering},
};

struct AuthorityState {
    next_request: AtomicU64,
}

/// The only issuer for one Database process and its Runtime request halves.
///
/// This type is concrete, has private fields, and is deliberately not Clone.
///
/// ```compile_fail
/// use saddle_core::DbPhysicalDispositionIssuer;
/// fn duplicate(issuer: DbPhysicalDispositionIssuer) {
///     let _copy = issuer.clone();
/// }
/// ```
///
/// ```compile_fail
/// use saddle_core::DbPhysicalDispositionIssuer;
/// fn forge() -> DbPhysicalDispositionIssuer {
///     DbPhysicalDispositionIssuer { state: unreachable!() }
/// }
/// ```
#[doc(hidden)]
pub struct DbPhysicalDispositionIssuer {
    state: Arc<AuthorityState>,
}

/// Database-side startup half. Database consumes it when the unique physical
/// process capability becomes live.
#[doc(hidden)]
pub struct DbPhysicalStartupHalf {
    state: Arc<AuthorityState>,
}

/// Runtime-side issuer retained next to the process Admission authority.
#[doc(hidden)]
pub struct DbPhysicalRequestIssuer {
    state: Arc<AuthorityState>,
}

/// Database-only process capability obtained by consuming the startup half.
#[doc(hidden)]
pub struct DbPhysicalProcessCapability {
    state: Arc<AuthorityState>,
}

/// Runtime half retained beside one admitted request/account generation.
#[doc(hidden)]
pub struct DbPhysicalRequestHalf {
    state: Arc<AuthorityState>,
    request: u64,
}

/// Database half moved with the same request into physical execution.
#[doc(hidden)]
pub struct DbPhysicalExecutionHalf {
    state: Arc<AuthorityState>,
    request: u64,
}

/// Database-owned proof created only after physical return or synchronous
/// poison-discard has completed. `T` remains indivisible from that outcome.
#[doc(hidden)]
pub struct DbPhysicalDispositionOwner<T> {
    state: Arc<AuthorityState>,
    request: u64,
    disposition: DbPhysicalDisposition,
    value: T,
}

/// Same-process, same-request sealed receipt consumed by Runtime before it
/// completes Admission accounting.
///
/// ```compile_fail
/// use saddle_core::DbPhysicalDispositionReceipt;
/// fn replay<T>(receipt: DbPhysicalDispositionReceipt<T>) {
///     let _first = receipt.into_outcome();
///     let _second = receipt.into_outcome();
/// }
/// ```
#[doc(hidden)]
pub struct DbPhysicalDispositionReceipt<T> {
    disposition: DbPhysicalDisposition,
    value: T,
}

/// Sealed proof that one paired request/execution authority was released
/// before any Database operation or physical acquisition began. `T` remains
/// linear with the terminal proof and the receipt is deliberately non-Clone.
///
/// ```compile_fail
/// use saddle_core::DbRequestNotUsedReceipt;
/// fn replay<T>(receipt: DbRequestNotUsedReceipt<T>) {
///     let _first = receipt.into_value();
///     let _second = receipt.into_value();
/// }
/// ```
#[doc(hidden)]
pub struct DbRequestNotUsedReceipt<T> {
    value: T,
}

/// Physical outcome carried by the sealed receipt. It has no default or
/// catch-all state.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum DbPhysicalDisposition {
    Returned,
    Discarded,
}

impl DbPhysicalDispositionIssuer {
    /// Creates one process authority. No caller-provided identity, digest,
    /// generation, or numeric seed participates in issuance.
    #[doc(hidden)]
    pub fn issue() -> Self {
        Self {
            state: Arc::new(AuthorityState {
                next_request: AtomicU64::new(1),
            }),
        }
    }

    /// Irreversibly splits the authority across Database startup and Runtime.
    #[doc(hidden)]
    pub fn into_startup_and_request_issuer(
        self,
    ) -> (DbPhysicalStartupHalf, DbPhysicalRequestIssuer) {
        (
            DbPhysicalStartupHalf {
                state: Arc::clone(&self.state),
            },
            DbPhysicalRequestIssuer { state: self.state },
        )
    }
}

impl DbPhysicalStartupHalf {
    /// Database consumes the startup half when its unique managed process
    /// owner is installed. No second process capability can be produced.
    #[doc(hidden)]
    pub fn into_process_capability(self) -> DbPhysicalProcessCapability {
        DbPhysicalProcessCapability { state: self.state }
    }
}

impl DbPhysicalRequestIssuer {
    /// Runtime calls this exactly while it moves the admitted DbRequestPermit
    /// and account generation into the managed request owner.
    #[doc(hidden)]
    pub fn issue_request(&self) -> Option<(DbPhysicalRequestHalf, DbPhysicalExecutionHalf)> {
        let request = self
            .state
            .next_request
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
                current.checked_add(1)
            })
            .ok()?;
        Some((
            DbPhysicalRequestHalf {
                state: Arc::clone(&self.state),
                request,
            },
            DbPhysicalExecutionHalf {
                state: Arc::clone(&self.state),
                request,
            },
        ))
    }
}

impl DbPhysicalProcessCapability {
    /// Seals a value after Database has completed a normal physical return.
    #[doc(hidden)]
    pub fn connection_returned<T>(
        &self,
        execution: DbPhysicalExecutionHalf,
        value: T,
    ) -> Result<DbPhysicalDispositionOwner<T>, (DbPhysicalExecutionHalf, T)> {
        self.seal(execution, DbPhysicalDisposition::Returned, value)
    }

    /// Seals a value after Database has synchronously detached and discarded
    /// a poisoned physical connection.
    #[doc(hidden)]
    pub fn connection_discarded<T>(
        &self,
        execution: DbPhysicalExecutionHalf,
        value: T,
    ) -> Result<DbPhysicalDispositionOwner<T>, (DbPhysicalExecutionHalf, T)> {
        self.seal(execution, DbPhysicalDisposition::Discarded, value)
    }

    fn seal<T>(
        &self,
        execution: DbPhysicalExecutionHalf,
        disposition: DbPhysicalDisposition,
        value: T,
    ) -> Result<DbPhysicalDispositionOwner<T>, (DbPhysicalExecutionHalf, T)> {
        if !Arc::ptr_eq(&self.state, &execution.state) {
            return Err((execution, value));
        }
        Ok(DbPhysicalDispositionOwner {
            state: execution.state,
            request: execution.request,
            disposition,
            value,
        })
    }
}

/// Consumes Database's physical owner and Runtime's admitted request half.
/// A foreign process/request returns every owner and `T` unchanged for the
/// original pairing; success leaves only the sealed receipt.
#[doc(hidden)]
pub fn pair_db_physical_disposition<T>(
    physical: DbPhysicalDispositionOwner<T>,
    request: DbPhysicalRequestHalf,
) -> Result<DbPhysicalDispositionReceipt<T>, (DbPhysicalDispositionOwner<T>, DbPhysicalRequestHalf)>
{
    if !Arc::ptr_eq(&physical.state, &request.state) || physical.request != request.request {
        return Err((physical, request));
    }
    Ok(DbPhysicalDispositionReceipt {
        disposition: physical.disposition,
        value: physical.value,
    })
}

impl<T> DbPhysicalDispositionReceipt<T> {
    /// Runtime consumes the sealed receipt while finishing the original
    /// Admission request. Neither the disposition nor `T` can be replayed.
    #[doc(hidden)]
    pub fn into_outcome(self) -> (DbPhysicalDisposition, T) {
        (self.disposition, self.value)
    }
}

/// Consumes both untouched halves of one request and seals the third terminal:
/// no Database operation or physical connection acquisition occurred. Crossed
/// halves return every input unchanged for the original pairing.
#[doc(hidden)]
pub fn seal_db_request_not_used<T>(
    request: DbPhysicalRequestHalf,
    execution: DbPhysicalExecutionHalf,
    value: T,
) -> Result<DbRequestNotUsedReceipt<T>, (DbPhysicalRequestHalf, DbPhysicalExecutionHalf, T)> {
    if !Arc::ptr_eq(&request.state, &execution.state) || request.request != execution.request {
        return Err((request, execution, value));
    }
    Ok(DbRequestNotUsedReceipt { value })
}

impl<T> DbRequestNotUsedReceipt<T> {
    /// Consumes the sealed terminal exactly once and releases the preserved
    /// value to Runtime for Admission credit finalization.
    #[doc(hidden)]
    pub fn into_value(self) -> T {
        self.value
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn authority() -> (DbPhysicalProcessCapability, DbPhysicalRequestIssuer) {
        let (startup, requests) =
            DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
        (startup.into_process_capability(), requests)
    }

    #[test]
    fn same_process_and_request_seal_returned_and_discarded() {
        let (process, requests) = authority();
        let (request, execution) = requests.issue_request().unwrap();
        let physical = match process.connection_returned(execution, "query") {
            Ok(value) => value,
            Err(_) => panic!("same process rejected"),
        };
        let receipt = match pair_db_physical_disposition(physical, request) {
            Ok(value) => value,
            Err(_) => panic!("same request rejected"),
        };
        assert_eq!(
            receipt.into_outcome(),
            (DbPhysicalDisposition::Returned, "query")
        );

        let (request, execution) = requests.issue_request().unwrap();
        let physical = match process.connection_discarded(execution, "write") {
            Ok(value) => value,
            Err(_) => panic!("same process rejected"),
        };
        let receipt = match pair_db_physical_disposition(physical, request) {
            Ok(value) => value,
            Err(_) => panic!("same request rejected"),
        };
        assert_eq!(
            receipt.into_outcome(),
            (DbPhysicalDisposition::Discarded, "write")
        );
    }

    #[test]
    fn foreign_process_and_crossed_request_return_all_owners() {
        let (first_process, first_requests) = authority();
        let (second_process, second_requests) = authority();
        let (first_request, first_execution) = first_requests.issue_request().unwrap();
        let (second_request, second_execution) = second_requests.issue_request().unwrap();

        let (first_execution, value) = match second_process.connection_returned(first_execution, 11)
        {
            Ok(_) => panic!("foreign process accepted"),
            Err(owners) => owners,
        };
        let first_physical = match first_process.connection_returned(first_execution, value) {
            Ok(value) => value,
            Err(_) => panic!("original process rejected"),
        };
        let second_physical = match second_process.connection_discarded(second_execution, 22) {
            Ok(value) => value,
            Err(_) => panic!("original process rejected"),
        };

        let (first_physical, second_request) =
            match pair_db_physical_disposition(first_physical, second_request) {
                Ok(_) => panic!("crossed request accepted"),
                Err(owners) => owners,
            };
        let (second_physical, first_request) =
            match pair_db_physical_disposition(second_physical, first_request) {
                Ok(_) => panic!("crossed request accepted"),
                Err(owners) => owners,
            };
        let first = match pair_db_physical_disposition(first_physical, first_request) {
            Ok(value) => value,
            Err(_) => panic!("original request rejected"),
        };
        assert_eq!(first.into_outcome(), (DbPhysicalDisposition::Returned, 11));
        let second = match pair_db_physical_disposition(second_physical, second_request) {
            Ok(value) => value,
            Err(_) => panic!("original request rejected"),
        };
        assert_eq!(
            second.into_outcome(),
            (DbPhysicalDisposition::Discarded, 22)
        );
    }

    #[test]
    fn untouched_request_seals_not_used_and_crossed_halves_retry() {
        let (_first_process, first_requests) = authority();
        let (_second_process, second_requests) = authority();
        let (first_request, first_execution) = first_requests.issue_request().unwrap();
        let (second_request, second_execution) = second_requests.issue_request().unwrap();

        let (first_request, second_execution, first_value) =
            match seal_db_request_not_used(first_request, second_execution, "first") {
                Ok(_) => panic!("crossed request accepted"),
                Err(owners) => owners,
            };
        let (second_request, first_execution, second_value) =
            match seal_db_request_not_used(second_request, first_execution, "second") {
                Ok(_) => panic!("crossed request accepted"),
                Err(owners) => owners,
            };
        let first = seal_db_request_not_used(first_request, first_execution, first_value)
            .unwrap_or_else(|_| panic!("original request rejected"));
        let second = seal_db_request_not_used(second_request, second_execution, second_value)
            .unwrap_or_else(|_| panic!("original request rejected"));
        assert_eq!(first.into_value(), "first");
        assert_eq!(second.into_value(), "second");
    }
}