hibana 0.5.2

//! Port abstraction for transport layer access.
//!
//! Port provides lightweight access to transport Tx/Rx handles,
//! scratch buffers, tap rings, and state tables.

use core::{
    cell::UnsafeCell,
    marker::PhantomData,
    task::{Context, Poll},
};

use super::tables::{LoopDisposition, LoopTable, RouteTable};
use crate::{
    control::types::{Lane, RendezvousId},
    endpoint::kernel::FrontierScratchLayout,
    global::const_dsl::ScopeId,
    observe::core::{TapRing, emit},
    policy_runtime::{self, PolicySlot},
    runtime::config::Clock,
    transport::{FrameLabelMask, Transport, TransportEvent, TransportEventKind, TransportMetrics},
};

const ROUTE_HINT_SLOTS: usize = u8::MAX as usize + 1;

#[derive(Clone, Copy)]
struct RouteHintQueue {
    present_mask: FrameLabelMask,
}

impl RouteHintQueue {
    #[cfg(test)]
    const fn new() -> Self {
        Self {
            present_mask: FrameLabelMask::EMPTY,
        }
    }

    const fn from_mask(present_mask: FrameLabelMask) -> Self {
        Self { present_mask }
    }

    fn push(&mut self, frame_label: u8) -> bool {
        if self.present_mask.contains_frame_label(frame_label) {
            return false;
        }
        self.present_mask.insert_frame_label(frame_label);
        true
    }

    fn take_matching<F>(&mut self, matches: F) -> Option<u8>
    where
        F: FnMut(u8) -> bool,
    {
        self.present_mask.take_matching(matches)
    }

    #[cfg(test)]
    fn has_matching<F>(&self, matches: F) -> bool
    where
        F: FnMut(u8) -> bool,
    {
        self.present_mask.has_matching(matches)
    }

    #[inline]
    fn has_any_frame_label_in_mask(&self, frame_label_mask: FrameLabelMask) -> bool {
        self.present_mask.intersects(frame_label_mask)
    }

    fn take_from_frame_label_mask(&mut self, frame_label_mask: FrameLabelMask) -> Option<u8> {
        self.take_matching(|frame_label| frame_label_mask.contains_frame_label(frame_label))
    }

    fn drain_from_transport<'a, T: Transport>(&mut self, transport: &'a T, rx: &'a T::Rx<'a>) {
        let mut budget = ROUTE_HINT_SLOTS;
        while budget > 0 {
            let frame_label = match transport.recv_frame_hint(rx) {
                Some(frame_label) => frame_label.raw(),
                None => break,
            };
            if !self.push(frame_label) {
                break;
            }
            budget -= 1;
        }
    }

    #[inline]
    fn clear(&mut self) {
        self.present_mask = FrameLabelMask::EMPTY;
    }
}

/// Lightweight port describing how an endpoint reaches the transport.
///
/// The port is intentionally `!Send`/`!Sync` (via the hidden `PhantomData`) so
/// that transport handles remain affine. Endpoint methods keep the mutable
/// borrow for the duration of a single `.await`, but exclusivity is preserved
/// because owning endpoints themselves are affine.
pub(crate) struct Port<
    'r,
    T: Transport,
    E: crate::control::cap::mint::EpochTable = crate::control::cap::mint::EpochTbl,
> {
    transport: &'r T,
    tx: UnsafeCell<T::Tx<'r>>,
    rx: UnsafeCell<T::Rx<'r>>,
    slab: *mut [u8],
    image_frontier: *const u32,
    scratch_reserved_bytes: *const u32,
    endpoint_leases: *const super::core::EndpointLeaseSlot,
    endpoint_lease_capacity: super::core::EndpointLeaseId,
    scratch_marker: PhantomData<&'r mut [u8]>,
    pub lane: Lane,
    role: u8,
    role_count: u8,
    rv_id: RendezvousId,
    _no_send_sync: PhantomData<*mut ()>,
    tap: *const TapRing<'static>,
    tap_marker: PhantomData<&'r TapRing<'r>>,
    clock: &'r dyn Clock,
    loops: *const LoopTable,
    loops_marker: PhantomData<&'r LoopTable>,
    routes: *const RouteTable,
    routes_marker: PhantomData<&'r RouteTable>,
    _epoch: PhantomData<E>,
}

impl<'r, T: Transport, E: crate::control::cap::mint::EpochTable + 'r> Port<'r, T, E> {
    #[inline(always)]
    const fn align_up(value: usize, align: usize) -> usize {
        let mask = align.saturating_sub(1);
        (value + mask) & !mask
    }

    #[inline(always)]
    const fn frontier_scratch_align() -> usize {
        FrontierScratchLayout::new(0, 0, 0).total_align()
    }

    #[inline]
    fn sync_pending_route_frame_hint_lane_masks(
        &self,
        before: FrameLabelMask,
        after: FrameLabelMask,
    ) {
        if before != after {
            self.route_table()
                .update_pending_frame_hint_mask_for_lane(self.lane, before, after);
        }
    }

    #[inline]
    fn route_hints_from_table(&self) -> RouteHintQueue {
        RouteHintQueue::from_mask(
            self.route_table()
                .pending_frame_hint_mask_for_lane(self.lane),
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new<'tap>(
        transport: &'r T,
        tap: &'tap TapRing<'tap>,
        clock: &'tap dyn Clock,
        loops: &'tap LoopTable,
        routes: &'tap RouteTable,
        slab: *mut [u8],
        image_frontier: *const u32,
        scratch_reserved_bytes: *const u32,
        endpoint_leases: *const super::core::EndpointLeaseSlot,
        endpoint_lease_capacity: super::core::EndpointLeaseId,
        lane: Lane,
        role: u8,
        role_count: u8,
        rv_id: RendezvousId,
        tx: T::Tx<'r>,
        rx: T::Rx<'r>,
    ) -> Self
    where
        'tap: 'r,
    {
        #[cfg(all(not(test), not(feature = "std")))]
        {
            let _ = tap;
            let _ = clock;
        }
        Self {
            transport,
            tx: UnsafeCell::new(tx),
            rx: UnsafeCell::new(rx),
            slab,
            image_frontier,
            scratch_reserved_bytes,
            endpoint_leases,
            endpoint_lease_capacity,
            scratch_marker: PhantomData,
            lane,
            role,
            role_count,
            rv_id,
            _no_send_sync: PhantomData,
            tap: (tap as *const TapRing<'tap>).cast::<TapRing<'static>>(),
            tap_marker: PhantomData,
            clock,
            loops: loops as *const LoopTable,
            loops_marker: PhantomData,
            routes: routes as *const RouteTable,
            routes_marker: PhantomData,
            _epoch: PhantomData,
        }
    }

    #[inline]
    fn slab_ptr_and_len(&self) -> (*mut u8, usize) {
        unsafe {
            let slab = &mut *self.slab;
            (slab.as_mut_ptr(), slab.len())
        }
    }

    #[inline]
    fn endpoint_storage_floor(&self) -> usize {
        let (_, slab_len) = self.slab_ptr_and_len();
        let mut floor = slab_len;
        let mut idx = 0usize;
        while idx < usize::from(self.endpoint_lease_capacity) {
            let slot = unsafe { &*self.endpoint_leases.add(idx) };
            if slot.occupied && slot.len != 0 && (slot.offset as usize) < floor {
                floor = slot.offset as usize;
            }
            idx += 1;
        }
        floor
    }

    pub(crate) fn transport(&self) -> &'r T {
        self.transport
    }

    #[inline]
    pub(crate) fn loop_table(&self) -> &LoopTable {
        unsafe { &*self.loops }
    }

    #[inline]
    pub(crate) fn route_table(&self) -> &RouteTable {
        unsafe { &*self.routes }
    }

    #[inline]
    pub(crate) fn record_loop_decision(&self, idx: u8, disposition: LoopDisposition) -> u16 {
        self.loop_table()
            .record(self.lane, self.role, idx, disposition)
    }

    #[inline]
    pub(crate) fn ack_loop_decision(&self, idx: u8, role: u8) {
        self.loop_table().acknowledge(self.lane, role, idx);
    }

    #[inline]
    pub(crate) fn record_route_decision(&self, scope: ScopeId, arm: u8) -> u16 {
        self.route_table()
            .record_with_role_count(self.lane, self.role_count, self.role, scope, arm)
    }

    #[inline]
    pub(crate) fn poll_route_decision(
        &self,
        scope: ScopeId,
        role: u8,
        cx: &mut Context<'_>,
    ) -> Poll<u8> {
        self.route_table()
            .poll_with_role_count(self.lane, self.role_count, role, scope, cx)
    }

    #[inline]
    pub(crate) fn ack_route_decision(&self, scope: ScopeId, role: u8) -> Option<u8> {
        self.route_table()
            .acknowledge_with_role_count(self.lane, self.role_count, role, scope)
    }

    #[inline]
    pub(crate) fn peek_route_decision(&self, scope: ScopeId, role: u8) -> Option<u8> {
        self.route_table()
            .peek_with_role_count(self.lane, self.role_count, role, scope)
    }

    #[inline]
    pub(crate) fn has_pending_route_decision_for_lane(
        &self,
        scope: ScopeId,
        role: u8,
        target_lane: Lane,
    ) -> bool {
        self.route_table().has_pending_lane_with_role_count(
            self.role_count,
            role,
            scope,
            target_lane,
        )
    }

    #[inline]
    pub(crate) fn route_change_epoch(&self) -> u16 {
        self.route_table().change_epoch()
    }

    #[cfg(test)]
    #[inline]
    pub(crate) fn has_route_hint_matching<F>(&self, matches: F) -> bool
    where
        F: FnMut(u8) -> bool,
    {
        let mut hints = self.route_hints_from_table();
        let before = hints.present_mask;
        let rx = unsafe { &*self.rx.get() };
        hints.drain_from_transport(self.transport(), rx);
        self.sync_pending_route_frame_hint_lane_masks(before, hints.present_mask);
        hints.has_matching(matches)
    }

    #[inline]
    pub(crate) fn has_route_hint_for_frame_label_mask(
        &self,
        frame_label_mask: FrameLabelMask,
        drain_transport_hints: bool,
    ) -> bool {
        let mut hints = self.route_hints_from_table();
        let before = hints.present_mask;
        if drain_transport_hints {
            let rx = unsafe { &*self.rx.get() };
            hints.drain_from_transport(self.transport(), rx);
        }
        self.sync_pending_route_frame_hint_lane_masks(before, hints.present_mask);
        hints.has_any_frame_label_in_mask(frame_label_mask)
    }

    #[inline]
    pub(crate) fn has_pending_route_hint_for_lane(
        &self,
        frame_label_mask: FrameLabelMask,
        target_lane: Lane,
        drain_transport_hints: bool,
    ) -> bool {
        let mut hints = self.route_hints_from_table();
        let before = hints.present_mask;
        if drain_transport_hints {
            let rx = unsafe { &*self.rx.get() };
            hints.drain_from_transport(self.transport(), rx);
        }
        self.sync_pending_route_frame_hint_lane_masks(before, hints.present_mask);
        self.route_table()
            .has_pending_frame_hint_for_lane(target_lane, frame_label_mask)
    }

    #[inline]
    pub(crate) fn take_route_hint_for_frame_label_mask(
        &self,
        frame_label_mask: FrameLabelMask,
        drain_transport_hints: bool,
    ) -> Option<u8> {
        let mut hints = self.route_hints_from_table();
        let before = hints.present_mask;
        if drain_transport_hints {
            let rx = unsafe { &*self.rx.get() };
            hints.drain_from_transport(self.transport(), rx);
        }
        let taken = hints.take_from_frame_label_mask(frame_label_mask);
        self.sync_pending_route_frame_hint_lane_masks(before, hints.present_mask);
        taken
    }

    #[inline]
    pub(crate) fn clear_route_hints(&self) {
        let mut hints = self.route_hints_from_table();
        let before = hints.present_mask;
        hints.clear();
        self.sync_pending_route_frame_hint_lane_masks(before, hints.present_mask);
    }

    #[inline]
    pub(crate) fn tx_ptr(&self) -> *mut T::Tx<'r> {
        self.tx.get()
    }

    #[inline]
    pub(crate) fn rx_ptr(&self) -> *mut T::Rx<'r> {
        self.rx.get()
    }

    #[inline]
    pub(crate) fn scratch_ptr(&self) -> *mut [u8] {
        let (ptr, _) = self.slab_ptr_and_len();
        let base = unsafe { *self.image_frontier } as usize;
        let reserved = unsafe { *self.scratch_reserved_bytes } as usize;
        let start = base.saturating_add(reserved);
        let end = self.endpoint_storage_floor();
        let len = end.saturating_sub(start);
        unsafe { core::ptr::slice_from_raw_parts_mut(ptr.add(start), len) }
    }

    #[inline]
    pub(crate) fn frontier_scratch_ptr(&self) -> *mut [u8] {
        let (ptr, _) = self.slab_ptr_and_len();
        let start = unsafe { *self.image_frontier } as usize;
        let reserved = unsafe { *self.scratch_reserved_bytes } as usize;
        let lease_floor = self.endpoint_storage_floor();
        let end = if reserved == 0 {
            lease_floor
        } else {
            core::cmp::min(start.saturating_add(reserved), lease_floor)
        };
        let scratch_start =
            core::cmp::min(Self::align_up(start, Self::frontier_scratch_align()), end);
        let len = end.saturating_sub(scratch_start);
        unsafe { core::ptr::slice_from_raw_parts_mut(ptr.add(scratch_start), len) }
    }

    #[inline]
    pub(crate) fn policy_digest(&self, slot: PolicySlot) -> u32 {
        let _ = slot;
        policy_runtime::POLICY_DIGEST_NONE
    }

    #[inline]
    pub(crate) fn tap(&self) -> &TapRing<'r> {
        unsafe { &*self.tap.cast::<TapRing<'r>>() }
    }

    #[inline]
    pub(crate) fn clock(&self) -> &dyn Clock {
        self.clock
    }

    #[inline]
    pub(crate) fn now32(&self) -> u32 {
        self.clock.now32()
    }

    #[inline]
    pub(crate) fn flush_transport_events(&self) -> Option<TransportEvent> {
        use crate::observe::events;
        let tap = self.tap();
        let clock = self.clock();
        let mut last_loss = None;
        let mut emit_event = |event: TransportEvent| {
            let (arg0, arg1) = event.encode_tap_args();
            if matches!(event.kind(), TransportEventKind::Loss) {
                last_loss = Some(event);
            }
            emit(tap, events::TransportEvent::new(clock.now32(), arg0, arg1));
        };
        self.transport.drain_events(&mut emit_event);
        let metrics_attrs = self.transport.metrics().attrs();
        let snapshot = crate::transport::TransportSnapshot::from_policy_attrs(&metrics_attrs);
        if let Some(payload) = snapshot.encode_tap_metrics() {
            let (arg0, arg1) = payload.primary();
            emit(
                tap,
                events::TransportMetrics::new(clock.now32(), arg0, arg1),
            );
            if let Some((ext0, ext1)) = payload.extension() {
                emit(
                    tap,
                    events::TransportMetricsExt::new(clock.now32(), ext0, ext1),
                );
            }
        }

        last_loss
    }

    #[inline]
    pub(crate) fn lane(&self) -> Lane {
        self.lane
    }

    #[inline]
    pub(crate) fn rv_id(&self) -> RendezvousId {
        self.rv_id
    }
}

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

    #[test]
    fn route_hint_unmatched_is_not_discarded_across_scope_selection() {
        let mut queue = RouteHintQueue::new();
        queue.push(41);
        queue.push(42);

        let first = queue.take_matching(|frame_label| frame_label == 99);
        assert_eq!(
            first, None,
            "non-matching take must not clear buffered hints"
        );

        let second = queue.take_matching(|frame_label| frame_label == 42);
        assert_eq!(
            second,
            Some(42),
            "later matching hint must remain available"
        );

        let third = queue.take_matching(|frame_label| frame_label == 41);
        assert_eq!(
            third,
            Some(41),
            "earlier unmatched hint must still be available after sibling selection"
        );
    }

    #[test]
    fn route_hint_queue_deduplicates_same_frame_label() {
        let mut queue = RouteHintQueue::new();
        queue.push(25);
        queue.push(25);
        queue.push(25);

        let first = queue.take_matching(|frame_label| frame_label == 25);
        assert_eq!(first, Some(25));

        let second = queue.take_matching(|frame_label| frame_label == 25);
        assert_eq!(
            second, None,
            "duplicate frame labels must be coalesced in queue"
        );
    }

    #[test]
    fn route_hint_has_matching_is_non_consuming() {
        let mut queue = RouteHintQueue::new();
        queue.push(25);
        queue.push(201);

        assert!(queue.has_matching(|frame_label| frame_label == 201));
        assert_eq!(
            queue.take_matching(|frame_label| frame_label == 201),
            Some(201)
        );
        assert_eq!(
            queue.take_matching(|frame_label| frame_label == 25),
            Some(25)
        );
    }
}