saddle-framework 0.3.26

//! Pre-allocation facts for the *same* constructors used by the entry loop.
//! This module issues no permission and does not change settlement order.
use super::*;
use saddle_runtime::request_task::{RequestTaskBorrowedFuture, RequestTaskContext};
use std::alloc::{Layout, LayoutError};

pub(super) struct NormalFactory<B, C, Dispatch> {
    pub(super) socket: TcpStream,
    pub(super) adapter: saddle_boundary::ingress::ProfuseGwListenerAdapter,
    pub(super) deployment: Option<C>,
    pub(super) business: Option<BusinessConfig<B>>,
    pub(super) ingress_token: Option<Arc<IngressToken>>,
    pub(super) dispatch: Arc<Dispatch>,
    pub(super) observer: saddle_observability::Observer,
    pub(super) admission: Option<saddle_runtime::profusegw::ProfuseGwAdmissionEvent>,
    pub(super) database: Option<Arc<ManagedDatabaseState>>,
    pub(super) diagnostic_handle: Option<saddle_observability::EmergencyDiagnosticHandle>,
}

impl<B, C, Dispatch> NormalFactory<B, C, Dispatch> {
    async fn body<DispatchFuture>(
        self,
        owner: &mut EntryRequestOwner,
        task: RequestTaskContext,
    ) -> std::result::Result<(), saddle_observability::FrameworkRequestFailure<std::io::Error>>
    where
        Dispatch: Fn(
            saddle_boundary::ingress::AcceptedIngress,
            C,
            BusinessConfig<B>,
            crate::database_capability::DatabaseRequest,
        ) -> DispatchFuture,
        DispatchFuture: Future<Output = Result<Vec<u8>>>,
    {
        serve_managed(
            self.socket,
            &self.adapter,
            self.deployment.expect("single deployment input"),
            self.business.expect("single business input"),
            self.ingress_token,
            &*self.dispatch,
            &self.observer,
            owner,
            task,
            self.admission.expect("legacy entry supplies its event"),
            self.database,
            self.diagnostic_handle,
        )
        .await
    }
}

// The existing Runtime ABI still allocates here. SC only exposes its concrete
// input before this call; R/S will replace the allocation authority later.
pub(super) fn boxed_normal_factory<B, C, Dispatch, DispatchFuture>(
    factory: NormalFactory<B, C, Dispatch>,
) -> impl for<'a> FnOnce(
    &'a mut EntryRequestOwner,
    RequestTaskContext,
) -> RequestTaskBorrowedFuture<'a, (), std::io::Error>
+ Send
+ 'static
where
    B: Send + Sync + 'static,
    C: Send + 'static,
    Dispatch: Fn(
            saddle_boundary::ingress::AcceptedIngress,
            C,
            BusinessConfig<B>,
            crate::database_capability::DatabaseRequest,
        ) -> DispatchFuture
        + Send
        + Sync
        + 'static,
    DispatchFuture: Future<Output = Result<Vec<u8>>> + Send + 'static,
{
    move |owner, context| Box::pin(factory.body(owner, context))
}

pub(super) struct RejectionFactory {
    pub(super) socket: TcpStream,
    pub(super) code: u16,
    pub(super) retained: CancelledDelivery,
}
impl RejectionFactory {
    pub(super) async fn body(
        self,
        context: RequestTaskContext,
    ) -> std::result::Result<
        RejectionResult,
        saddle_observability::FrameworkRequestFailure<std::io::Error>,
    > {
        let output = context.output();
        let delivery = reject_required(
            self.socket,
            self.code,
            context.scope().with_output(output.as_ref()),
            self.retained,
            output.clone(),
        )
        .await;
        Ok(RejectionResult { delivery, output })
    }
}

macro_rules! rejection_factory {
    ($factory:expr) => {{
        let factory = $factory;
        move |context| factory.body(context)
    }};
}
pub(super) use rejection_factory;

// Infers R without invoking the constructor, creating arguments or allocating R.
fn output_layout<A, R>(_: impl FnOnce(A) -> R) -> Layout {
    Layout::new::<R>()
}

/// Payload and current standard-library shared allocation, not allocator RSS.
#[derive(Clone, Copy, Debug)]
pub struct SharedLayout {
    pub payload: Layout,
    pub allocation: Layout,
}
pub(crate) fn shared_layout<T>() -> std::result::Result<SharedLayout, LayoutError> {
    let payload = Layout::new::<T>();
    let (allocation, _) = Layout::new::<[std::sync::atomic::AtomicUsize; 2]>().extend(payload)?;
    Ok(SharedLayout {
        payload,
        allocation: allocation.pad_to_align(),
    })
}

/// Service inputs only. R0 must add its wrapper/state/task and container
/// capacity/control allocations, not multiply these by active admission count.
#[derive(Debug)]
pub struct EntryConsumerLayouts {
    pub normal_runtime: saddle_runtime::request_task::storage::TaskLayouts,
    pub rejection_runtime: saddle_runtime::request_task::storage::TaskLayouts,
    pub normal_environment: Layout,
    pub normal_factory: Layout,
    pub normal_body: Layout,
    pub normal_result: Layout,
    pub normal_owner: Layout,
    pub dispatch_future: Layout,
    pub process_dispatch_cell: SharedLayout,
    pub owner_cell: SharedLayout,
    pub normal_ticket_entry: Layout,
    pub rejection_environment: Layout,
    pub rejection_factory: Layout,
    pub rejection_body: Layout,
    pub rejection_result: Layout,
    pub rejection_ticket_entry: Layout,
    pub cancellation_cell: SharedLayout,
    pub delivery_guard: Layout,
    pub database: DatabaseConsumerLayouts,
}

/// Handle sizes are already embedded in their owners/futures. Do not add them
/// again. Retention Vec backing storage is capacity-dependent and not included.
#[derive(Debug)]
pub struct DatabaseConsumerLayouts {
    pub state: SharedLayout,
    /// Separate Box allocation on compatibility (non-reserved) requests only.
    /// Formal reserved requests neither allocate nor reserve this object.
    pub legacy_diagnostics: Layout,
    pub cancellation_notify: SharedLayout,
    pub outbound_retentions: SharedLayout,
    pub outbound_retention_element: Layout,
    pub request_handle: Layout,
    pub completion_handle: Layout,
    pub encoding_handle: Layout,
}

/// No factory invocation, request allocation, admission or body poll occurs.
/// Use the same B/C/Dispatch/DispatchFuture monomorphization as the application.
#[doc(hidden)]
pub fn entry_consumer_layouts<B, C, Dispatch, DispatchFuture>()
-> std::result::Result<EntryConsumerLayouts, LayoutError>
where
    B: Send + Sync + 'static,
    C: Send + 'static,
    Dispatch: Fn(
            saddle_boundary::ingress::AcceptedIngress,
            C,
            BusinessConfig<B>,
            crate::database_capability::DatabaseRequest,
        ) -> DispatchFuture
        + Send
        + Sync
        + 'static,
    DispatchFuture: Future<Output = Result<Vec<u8>>> + Send + 'static,
{
    Ok(EntryConsumerLayouts {
        normal_runtime: saddle_runtime::request_task::storage::owned_layouts::<EntryRequestOwner, (), std::io::Error, _, _>(boxed_normal_factory::<B, C, Dispatch, DispatchFuture>)?,
        rejection_runtime: saddle_runtime::request_task::storage::pair_layouts(|factory: RejectionFactory| rejection_factory!(factory))?,
        normal_environment: Layout::new::<NormalFactory<B, C, Dispatch>>(),
        normal_factory: output_layout(boxed_normal_factory::<B, C, Dispatch, DispatchFuture>),
        normal_body: output_layout(
            |(factory, owner, context): (
                NormalFactory<B, C, Dispatch>,
                &'static mut EntryRequestOwner,
                RequestTaskContext,
            )| factory.body::<DispatchFuture>(owner, context),
        ),
        normal_result: Layout::new::<
            std::result::Result<(), saddle_observability::FrameworkRequestFailure<std::io::Error>>,
        >(),
        normal_owner: Layout::new::<EntryRequestOwner>(),
        dispatch_future: Layout::new::<DispatchFuture>(),
        process_dispatch_cell: shared_layout::<Dispatch>()?,
        owner_cell: shared_layout::<tokio::sync::Mutex<EntryRequestOwner>>()?,
        normal_ticket_entry: Layout::new::<(
            tokio::task::Id,
            saddle_runtime::request_task::RequestTaskOwnedJoin<EntryRequestOwner>,
        )>(),
        rejection_environment: Layout::new::<RejectionFactory>(),
        rejection_factory: output_layout(|factory: RejectionFactory| rejection_factory!(factory)),
        rejection_body: output_layout(
            |(factory, context): (RejectionFactory, RequestTaskContext)| factory.body(context),
        ),
        rejection_result: Layout::new::<
            std::result::Result<
                RejectionResult,
                saddle_observability::FrameworkRequestFailure<std::io::Error>,
            >,
        >(),
        rejection_ticket_entry: Layout::new::<(
            tokio::task::Id,
            (
                saddle_runtime::request_task::RequestTaskJoin,
                CancelledDelivery,
            ),
        )>(),
        cancellation_cell: shared_layout::<std::sync::Mutex<Option<RejectionResult>>>()?,
        delivery_guard: Layout::new::<DeliveryGuard<'static>>(),
        database: crate::database_capability::request_storage_layouts()?,
    })
}

/// Infer the application dispatch types without calling it. The reference is
/// only a type witness; no closure captures are read or allocated by this API.
#[doc(hidden)]
pub fn entry_consumer_layouts_for<B, C, Dispatch, DispatchFuture>(
    _: &Dispatch,
) -> std::result::Result<EntryConsumerLayouts, LayoutError>
where
    B: Send + Sync + 'static,
    C: Send + 'static,
    Dispatch: Fn(
            saddle_boundary::ingress::AcceptedIngress,
            C,
            BusinessConfig<B>,
            crate::database_capability::DatabaseRequest,
        ) -> DispatchFuture
        + Send
        + Sync
        + 'static,
    DispatchFuture: Future<Output = Result<Vec<u8>>> + Send + 'static,
{
    entry_consumer_layouts::<B, C, Dispatch, DispatchFuture>()
}