saddle-runtime 0.3.0-alpha.4

//! Alpha.1 bridge from a fixed, parsed Ingress owner into the 0.2 request domain.
//!
//! This is intentionally an execution seam, not a listener or protocol API.

use std::{
    future::Future,
    pin::Pin,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    },
    task::{Context, Poll},
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use tokio::{sync::oneshot, time::Instant};

use crate::{RequestLifecycle, request::RequestClaim};

/// Verified absolute deadline consumed by one managed request submission.
///
/// It is non-cloneable and exposes no raw duration or Tokio instant.
#[doc(hidden)]
pub struct AbsoluteDeadlineOwner {
    instant: Instant,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum AbsoluteDeadlineError {
    ClockBeforeUnixEpoch,
    Expired,
    OutOfRange,
}

/// Converts the protocol absolute deadline once, before Admission publication.
#[doc(hidden)]
pub fn verify_absolute_deadline(
    deadline_unix_ms: u64,
) -> Result<AbsoluteDeadlineOwner, AbsoluteDeadlineError> {
    let now_wall = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| AbsoluteDeadlineError::ClockBeforeUnixEpoch)?;
    let now_ms =
        u64::try_from(now_wall.as_millis()).map_err(|_| AbsoluteDeadlineError::OutOfRange)?;
    let remaining_ms = deadline_unix_ms
        .checked_sub(now_ms)
        .filter(|remaining| *remaining > 0)
        .ok_or(AbsoluteDeadlineError::Expired)?;
    let instant = Instant::now()
        .checked_add(Duration::from_millis(remaining_ms))
        .ok_or(AbsoluteDeadlineError::OutOfRange)?;
    Ok(AbsoluteDeadlineOwner { instant })
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum ManagedIngressRejectReason {
    RuntimeUnavailable,
    AtCapacity,
    Lifecycle,
}

/// Recoverable pre-publication rejection. The exact input and deadline owner
/// are returned; no task or request guard exists.
#[doc(hidden)]
pub struct ManagedIngressRejection<I> {
    input: I,
    deadline: AbsoluteDeadlineOwner,
    reason: ManagedIngressRejectReason,
}

impl<I> std::fmt::Debug for ManagedIngressRejection<I> {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ManagedIngressRejection")
            .field("reason", &self.reason)
            .finish_non_exhaustive()
    }
}

impl<I> ManagedIngressRejection<I> {
    pub fn into_parts(self) -> (I, AbsoluteDeadlineOwner, ManagedIngressRejectReason) {
        (self.input, self.deadline, self.reason)
    }
}

/// The one terminal emitted by the Runtime-owned request task.
#[derive(Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum ManagedIngressTerminal<O, E> {
    Completed(Result<O, E>),
    DeadlineExceeded,
    Panicked,
}

/// Consumable post-finalizer resource fact. `Zero` is signed only after the
/// request lifecycle guard has been returned and this bridge has no other
/// occupied request generation at the finalizer linearization point.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum ManagedIngressResourceState {
    Zero,
    NonZero,
}

/// Non-cloneable evidence emitted by the unique request finalizer.
#[doc(hidden)]
pub struct ManagedIngressFinalized<O, E> {
    terminal: ManagedIngressTerminal<O, E>,
    resources: ManagedIngressResourceState,
}

impl<O, E> ManagedIngressFinalized<O, E> {
    pub fn into_parts(self) -> (ManagedIngressTerminal<O, E>, ManagedIngressResourceState) {
        (self.terminal, self.resources)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub struct ManagedIngressFinalizerLost;

/// Failure from the supported one-shot composition seam. A submission
/// rejection is preserved exactly; a missing finalizer remains fail-closed.
#[doc(hidden)]
pub enum ManagedIngressCompositionFailure<R> {
    InvalidCapacity,
    Rejected(R),
    FinalizerLost(ManagedIngressFinalizerLost),
}

/// Failure from the fixed-ingress composition entry. Deadline conversion is
/// recoverable before Admission; later rejection preserves Runtime's exact
/// linear inputs through [`ManagedIngressRejection`].
#[doc(hidden)]
pub enum FixedIngressCompositionFailure<I> {
    InvalidDeadline {
        input: I,
        error: AbsoluteDeadlineError,
    },
    Runtime(ManagedIngressCompositionFailure<ManagedIngressRejection<I>>),
}

/// Supported Alpha.4 composition entry for one already-accepted, fixed
/// ingress owner.
///
/// Runtime alone converts the absolute deadline and performs the managed
/// submission. `execute` is the one request closure: protocol decoding,
/// Service handling and an external call may happen inside it, while the
/// ingress owner, lifecycle claim and fixed slot remain owned by the same
/// managed request. The result is observable only after the unique finalizer
/// has returned those Runtime resources.
#[doc(hidden)]
pub async fn execute_fixed_ingress_once<I, F, Fut, O, E>(
    capacity: usize,
    input: I,
    deadline_unix_ms: u64,
    execute: F,
) -> Result<ManagedIngressFinalized<O, E>, FixedIngressCompositionFailure<I>>
where
    I: Send + 'static,
    F: FnOnce(I) -> Fut + Send + 'static,
    Fut: Future<Output = Result<O, E>> + Send + 'static,
    O: Send + 'static,
    E: Send + 'static,
{
    let deadline = match verify_absolute_deadline(deadline_unix_ms) {
        Ok(deadline) => deadline,
        Err(error) => {
            return Err(FixedIngressCompositionFailure::InvalidDeadline { input, error });
        }
    };
    compose_managed_ingress_once(capacity, move |bridge| {
        bridge.try_submit(input, deadline, execute)
    })
    .await
    .map_err(FixedIngressCompositionFailure::Runtime)
}

#[cfg(test)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ManagedIngressResourceSnapshot {
    active: usize,
    occupied_slots: usize,
    terminals: usize,
}

struct Shared {
    requests: RequestLifecycle,
    slots: Box<[AtomicBool]>,
    active: AtomicUsize,
    terminals: AtomicUsize,
}

/// Fixed-capacity ingress bridge. It owns no listener and exposes no executor.
#[derive(Clone)]
#[doc(hidden)]
pub struct ManagedIngressBridge {
    shared: Arc<Shared>,
}

impl ManagedIngressBridge {
    pub(crate) fn new(requests: RequestLifecycle, capacity: usize) -> Option<Self> {
        if capacity == 0 {
            return None;
        }
        Some(Self {
            shared: Arc::new(Shared {
                requests,
                slots: (0..capacity).map(|_| AtomicBool::new(false)).collect(),
                active: AtomicUsize::new(0),
                terminals: AtomicUsize::new(0),
            }),
        })
    }

    /// Moves an owned, already parsed fixed-Ingress input into the managed
    /// request domain. Dropping the returned completion does not cancel work.
    #[allow(clippy::type_complexity)]
    pub fn try_submit<I, F, Fut, O, E>(
        &self,
        input: I,
        deadline: AbsoluteDeadlineOwner,
        execute: F,
    ) -> Result<ManagedIngressCompletion<O, E>, ManagedIngressRejection<I>>
    where
        I: Send + 'static,
        F: FnOnce(I) -> Fut + Send + 'static,
        Fut: Future<Output = Result<O, E>> + Send + 'static,
        O: Send + 'static,
        E: Send + 'static,
    {
        if tokio::runtime::Handle::try_current().is_err() {
            return Err(ManagedIngressRejection {
                input,
                deadline,
                reason: ManagedIngressRejectReason::RuntimeUnavailable,
            });
        }
        let Some(slot) = self.shared.slots.iter().position(|slot| {
            slot.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
        }) else {
            return Err(ManagedIngressRejection {
                input,
                deadline,
                reason: ManagedIngressRejectReason::AtCapacity,
            });
        };
        self.shared.active.fetch_add(1, Ordering::AcqRel);
        let claim = match self.shared.requests.try_claim() {
            Ok(claim) => claim,
            Err(_) => {
                self.shared.active.fetch_sub(1, Ordering::AcqRel);
                self.shared.slots[slot].store(false, Ordering::Release);
                return Err(ManagedIngressRejection {
                    input,
                    deadline,
                    reason: ManagedIngressRejectReason::Lifecycle,
                });
            }
        };
        let (sender, receiver) = oneshot::channel();
        let finalizer = RequestFinalizer {
            shared: Arc::clone(&self.shared),
            slot,
            lifecycle: Some(RequestClaim::publish(claim)),
            sender: Some(sender),
            finished: false,
        };
        tokio::spawn(async move {
            let terminal = match tokio::time::timeout_at(deadline.instant, execute(input)).await {
                Ok(result) => ManagedIngressTerminal::Completed(result),
                Err(_) => ManagedIngressTerminal::DeadlineExceeded,
            };
            finalizer.finish(terminal);
        });
        Ok(ManagedIngressCompletion { receiver })
    }

    #[cfg(test)]
    fn resource_snapshot(&self) -> ManagedIngressResourceSnapshot {
        ManagedIngressResourceSnapshot {
            active: self.shared.active.load(Ordering::Acquire),
            occupied_slots: self
                .shared
                .slots
                .iter()
                .filter(|slot| slot.load(Ordering::Acquire))
                .count(),
            terminals: self.shared.terminals.load(Ordering::Acquire),
        }
    }
}

/// Supported non-business composition seam for framework integration and
/// cross-component verification.
///
/// Runtime exclusively performs the Ready transition, invokes exactly one
/// submission closure, then drains the same 0.2 lifecycle before returning
/// the post-finalizer receipt. It is not a listener or production entry API.
#[doc(hidden)]
pub async fn compose_managed_ingress_once<F, O, E, R>(
    capacity: usize,
    submit: F,
) -> Result<ManagedIngressFinalized<O, E>, ManagedIngressCompositionFailure<R>>
where
    F: FnOnce(&ManagedIngressBridge) -> Result<ManagedIngressCompletion<O, E>, R>,
{
    let requests = RequestLifecycle::new();
    let Some(bridge) = ManagedIngressBridge::new(requests.clone(), capacity) else {
        return Err(ManagedIngressCompositionFailure::InvalidCapacity);
    };
    requests.mark_ready();
    let completion = match submit(&bridge) {
        Ok(completion) => completion,
        Err(rejection) => {
            requests.begin_draining();
            requests.wait_until_drained().await;
            requests.mark_stopped();
            return Err(ManagedIngressCompositionFailure::Rejected(rejection));
        }
    };
    let finalized = completion.await;
    requests.begin_draining();
    requests.wait_until_drained().await;
    requests.mark_stopped();
    finalized.map_err(ManagedIngressCompositionFailure::FinalizerLost)
}

struct RequestFinalizer<O, E> {
    shared: Arc<Shared>,
    slot: usize,
    lifecycle: Option<crate::RequestGuard>,
    sender: Option<oneshot::Sender<ManagedIngressFinalized<O, E>>>,
    finished: bool,
}

impl<O, E> RequestFinalizer<O, E> {
    fn finish(mut self, terminal: ManagedIngressTerminal<O, E>) {
        self.record(terminal);
    }

    fn record(&mut self, terminal: ManagedIngressTerminal<O, E>) {
        // This is the finalizer linearization sequence. The receipt cannot be
        // observed until the lifecycle and fixed-slot resources are returned.
        drop(self.lifecycle.take());
        let previous_active = self.shared.active.fetch_sub(1, Ordering::AcqRel);
        debug_assert!(previous_active > 0);
        self.shared.slots[self.slot].store(false, Ordering::Release);
        self.shared.terminals.fetch_add(1, Ordering::AcqRel);
        self.finished = true;
        let resources = if previous_active == 1 {
            ManagedIngressResourceState::Zero
        } else {
            ManagedIngressResourceState::NonZero
        };
        let _ = self
            .sender
            .take()
            .expect("terminal sender exists")
            .send(ManagedIngressFinalized {
                terminal,
                resources,
            });
    }
}

impl<O, E> Drop for RequestFinalizer<O, E> {
    fn drop(&mut self) {
        if !self.finished {
            self.record(ManagedIngressTerminal::Panicked);
        }
    }
}

/// Awaitable terminal receiver. Dropping it detaches only the response waiter;
/// the Runtime-owned task, finalizer and lifecycle guard continue to terminal.
#[doc(hidden)]
pub struct ManagedIngressCompletion<O, E> {
    receiver: oneshot::Receiver<ManagedIngressFinalized<O, E>>,
}

impl<O, E> Future for ManagedIngressCompletion<O, E> {
    type Output = Result<ManagedIngressFinalized<O, E>, ManagedIngressFinalizerLost>;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        match Pin::new(&mut self.receiver).poll(context) {
            Poll::Ready(Ok(finalized)) => Poll::Ready(Ok(finalized)),
            Poll::Ready(Err(_)) => Poll::Ready(Err(ManagedIngressFinalizerLost)),
            Poll::Pending => Poll::Pending,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        time::{Duration, SystemTime, UNIX_EPOCH},
    };

    use super::*;
    use tokio::net::{TcpListener, TcpStream};

    fn deadline(after: Duration) -> AbsoluteDeadlineOwner {
        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
        verify_absolute_deadline(u64::try_from((now + after).as_millis()).unwrap()).unwrap()
    }

    fn deadline_unix_ms(after: Duration) -> u64 {
        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
        u64::try_from((now + after).as_millis()).unwrap()
    }

    async fn socket_pair() -> (TcpStream, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let client = TcpStream::connect(address);
        let (client, server) = tokio::join!(client, listener.accept());
        (client.unwrap(), server.unwrap().0)
    }

    #[tokio::test]
    async fn fixed_http_ingress_keeps_handler_and_external_call_in_one_managed_closure() {
        let (client, server) = socket_pair().await;
        let external_calls = Arc::new(AtomicUsize::new(0));
        let observed = Arc::clone(&external_calls);

        let request = tokio::spawn(async move {
            match execute_fixed_ingress_once(
                1,
                server,
                deadline_unix_ms(Duration::from_secs(2)),
                move |socket| async move {
                    let ingress = socket.peer_addr().unwrap();
                    // This nested future represents the Service handler's
                    // real external call and remains inside this request.
                    async {
                        observed.fetch_add(1, Ordering::AcqRel);
                    }
                    .await;
                    Ok::<_, ()>(ingress)
                },
            )
            .await
            {
                Ok(finalized) => finalized,
                Err(_) => panic!("valid accepted ingress must complete"),
            }
        });

        let (terminal, resources) = request.await.unwrap().into_parts();
        assert_eq!(
            terminal,
            ManagedIngressTerminal::Completed(Ok(client.local_addr().unwrap()))
        );
        assert_eq!(resources, ManagedIngressResourceState::Zero);
        assert_eq!(external_calls.load(Ordering::Acquire), 1);
    }

    #[tokio::test]
    async fn fixed_ingress_invalid_deadline_returns_the_exact_socket_before_admission() {
        let (_client, server) = socket_pair().await;
        let result =
            execute_fixed_ingress_once(1, server, 0, |_socket| async { Ok::<(), ()>(()) }).await;
        let (input, error) = match result {
            Err(FixedIngressCompositionFailure::InvalidDeadline { input, error }) => (input, error),
            _ => panic!("expired deadline must reject before managed submission"),
        };
        assert_eq!(error, AbsoluteDeadlineError::Expired);
        drop(input);
    }

    async fn ready_bridge(capacity: usize) -> ManagedIngressBridge {
        let requests = RequestLifecycle::new();
        requests.mark_ready();
        ManagedIngressBridge::new(requests, capacity).unwrap()
    }

    #[tokio::test]
    async fn success_deadline_panic_and_detached_waiter_all_return_resources() {
        let bridge = ready_bridge(1).await;
        let (completed, completed_resources) = bridge
            .try_submit(7, deadline(Duration::from_secs(1)), |value| async move {
                Ok::<_, ()>(value + 1)
            })
            .unwrap()
            .await
            .unwrap()
            .into_parts();
        assert_eq!(completed, ManagedIngressTerminal::Completed(Ok(8)));
        assert_eq!(completed_resources, ManagedIngressResourceState::Zero);

        let (expired, expired_resources) = bridge
            .try_submit((), deadline(Duration::from_millis(5)), |_| async {
                tokio::time::sleep(Duration::from_secs(1)).await;
                Ok::<_, ()>(())
            })
            .unwrap()
            .await
            .unwrap()
            .into_parts();
        assert_eq!(expired, ManagedIngressTerminal::DeadlineExceeded);
        assert_eq!(expired_resources, ManagedIngressResourceState::Zero);

        let (panicked, panicked_resources) = bridge
            .try_submit((), deadline(Duration::from_secs(1)), |_| async {
                panic!("managed panic");
                #[allow(unreachable_code)]
                Ok::<(), ()>(())
            })
            .unwrap()
            .await
            .unwrap()
            .into_parts();
        assert_eq!(panicked, ManagedIngressTerminal::Panicked);
        assert_eq!(panicked_resources, ManagedIngressResourceState::Zero);

        let drops = Arc::new(AtomicUsize::new(0));
        struct Input(Arc<AtomicUsize>);
        impl Drop for Input {
            fn drop(&mut self) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }
        let completion = bridge
            .try_submit(
                Input(Arc::clone(&drops)),
                deadline(Duration::from_secs(1)),
                |input| async move {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    drop(input);
                    Ok::<_, ()>(())
                },
            )
            .unwrap();
        drop(completion);
        tokio::time::sleep(Duration::from_millis(30)).await;
        assert_eq!(drops.load(Ordering::SeqCst), 1);
        let snapshot = bridge.resource_snapshot();
        assert_eq!(snapshot.active, 0);
        assert_eq!(snapshot.occupied_slots, 0);
        assert_eq!(snapshot.terminals, 4);
    }

    #[tokio::test]
    async fn fixed_capacity_rejection_recovers_input_and_deadline_without_a_task() {
        let bridge = ready_bridge(1).await;
        let release = Arc::new(tokio::sync::Notify::new());
        let entered = Arc::new(tokio::sync::Notify::new());
        let first = bridge
            .try_submit((), deadline(Duration::from_secs(1)), {
                let release = Arc::clone(&release);
                let entered = Arc::clone(&entered);
                move |_| async move {
                    entered.notify_one();
                    release.notified().await;
                    Ok::<(), ()>(())
                }
            })
            .unwrap();
        entered.notified().await;

        let rejection = match bridge.try_submit(17, deadline(Duration::from_secs(1)), |_| async {
            Ok::<_, ()>(())
        }) {
            Ok(_) => panic!("second fixed slot must be rejected"),
            Err(rejection) => rejection,
        };
        let (input, recovered_deadline, reason) = rejection.into_parts();
        assert_eq!(input, 17);
        assert_eq!(reason, ManagedIngressRejectReason::AtCapacity);
        assert_eq!(bridge.resource_snapshot().active, 1);

        release.notify_one();
        let (first_terminal, first_resources) = first.await.unwrap().into_parts();
        assert_eq!(first_terminal, ManagedIngressTerminal::Completed(Ok(())));
        assert_eq!(first_resources, ManagedIngressResourceState::Zero);
        let (retry, retry_resources) = bridge
            .try_submit(input, recovered_deadline, |_| async { Ok::<_, ()>(()) })
            .unwrap()
            .await
            .unwrap()
            .into_parts();
        assert_eq!(retry, ManagedIngressTerminal::Completed(Ok(())));
        assert_eq!(retry_resources, ManagedIngressResourceState::Zero);
        let snapshot = bridge.resource_snapshot();
        assert_eq!(snapshot.active, 0);
        assert_eq!(snapshot.occupied_slots, 0);
        assert_eq!(snapshot.terminals, 2);
    }

    #[tokio::test]
    async fn finalized_receipt_distinguishes_nonzero_then_zero() {
        let bridge = ready_bridge(2).await;
        let release_first = Arc::new(tokio::sync::Notify::new());
        let release_second = Arc::new(tokio::sync::Notify::new());
        let first = bridge
            .try_submit((), deadline(Duration::from_secs(1)), {
                let release = Arc::clone(&release_first);
                move |_| async move {
                    release.notified().await;
                    Ok::<(), ()>(())
                }
            })
            .unwrap();
        let second = bridge
            .try_submit((), deadline(Duration::from_secs(1)), {
                let release = Arc::clone(&release_second);
                move |_| async move {
                    release.notified().await;
                    Ok::<(), ()>(())
                }
            })
            .unwrap();

        release_first.notify_one();
        let (_, first_resources) = first.await.unwrap().into_parts();
        assert_eq!(first_resources, ManagedIngressResourceState::NonZero);
        release_second.notify_one();
        let (_, second_resources) = second.await.unwrap().into_parts();
        assert_eq!(second_resources, ManagedIngressResourceState::Zero);
        assert_eq!(bridge.resource_snapshot().active, 0);
        assert_eq!(bridge.resource_snapshot().occupied_slots, 0);
    }

    #[test]
    fn application_issues_only_one_fixed_bridge() {
        let application = crate::Application::new();
        assert!(application.managed_ingress_bridge(0).is_none());
        assert!(application.managed_ingress_bridge(1).is_some());
        assert!(application.managed_ingress_bridge(1).is_none());
    }
}