liminal-sdk 0.19.0

//! The adapter's owner-visible signal queue and sink, platform-neutral so its
//! bound and its loss accounting are pinned natively.
//!
//! Signals are queued under a short borrow and drained to the sink outside
//! every seam borrow, so a sink handler may call back into the adapter
//! (send/close) without deadlocking the seam; a signal produced during such a
//! nested call is delivered by the outer drain loop. The queue only grows
//! while a sink handler is on the stack, so it is bounded at
//! [`MAX_PENDING_SIGNALS`]: a handler that emits more than that before
//! returning loses the excess, and the owner is told by name how many.

use alloc::boxed::Box;
use alloc::collections::VecDeque;
use alloc::string::String;
use core::cell::{Cell, RefCell};

use super::super::core::DriverOutput;

#[cfg(test)]
#[path = "signal_port_tests.rs"]
mod tests;

/// Most signals held while a sink handler is running. A handler that
/// re-enters the adapter emits one signal per call; this allows that many
/// nested calls in one handler, which is far beyond the one-outstanding
/// exchange rule's legitimate shape, and holds at most this many small
/// enums.
pub const MAX_PENDING_SIGNALS: usize = 256;

/// Owner-installed sink receiving every adapter signal, in order.
///
/// The wasm leg is single-threaded and event-driven: the sink runs on the
/// browser event loop, directly from the socket callback that produced the
/// signal. It deliberately does not require `Send + Sync` (F5, following the
/// beamr-wasm precedent).
pub type AdapterSignalSink = Box<dyn FnMut(AdapterSignal)>;

/// One signal delivered to the owner's sink.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AdapterSignal {
    /// A decision output of the shared driver.
    Output(DriverOutput),
    /// A typed adapter fault outside the driver's closed output set.
    Fault(AdapterFault),
}

/// Typed faults the shim itself can observe.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AdapterFault {
    /// The single-threaded driver seam was re-entered while borrowed. This is
    /// structurally unreachable under browser event-loop semantics and is
    /// reported instead of being silently dropped.
    SeamReentered,
    /// Executing a driver-emitted command on the browser socket failed.
    CommandExecutionFailed {
        /// Human-readable detail from the JavaScript boundary.
        description: String,
    },
    /// Signals were lost since the last report: `overflowed` because the
    /// queue already held [`MAX_PENDING_SIGNALS`] while a sink handler ran,
    /// `unqueued` because the queue itself was borrowed when they were
    /// emitted. Delivered once the queue drains; the stream before it is
    /// incomplete, so the owner treats this adapter as failed.
    SignalsLost {
        /// Signals refused by the bound.
        overflowed: u64,
        /// Signals that could not be queued at all.
        unqueued: u64,
    },
}

/// The signal port's counters.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct SignalPortStats {
    /// Most signals held at once.
    pub peak_pending: usize,
    /// Signals refused by [`MAX_PENDING_SIGNALS`].
    pub overflowed: u64,
    /// Signals that could not be queued because the queue was borrowed.
    pub unqueued: u64,
    /// Drains that found the queue borrowed and left its signals for the
    /// drain already running.
    pub drain_conflicts: u64,
}

/// Signal queue plus sink.
pub(super) struct SignalPort {
    queue: RefCell<VecDeque<AdapterSignal>>,
    sink: RefCell<AdapterSignalSink>,
    stats: Cell<SignalPortStats>,
    /// Losses not yet reported to the sink.
    owed: Cell<(u64, u64)>,
}

impl SignalPort {
    pub(super) fn new(sink: AdapterSignalSink) -> Self {
        Self {
            queue: RefCell::new(VecDeque::new()),
            sink: RefCell::new(sink),
            stats: Cell::new(SignalPortStats::default()),
            owed: Cell::new((0, 0)),
        }
    }

    /// The counters as they stand.
    pub(super) fn stats(&self) -> SignalPortStats {
        self.stats.get()
    }

    fn update(&self, change: impl FnOnce(&mut SignalPortStats)) {
        let mut stats = self.stats.get();
        change(&mut stats);
        self.stats.set(stats);
    }

    /// Queues one signal and drains the queue to the sink if it is free.
    ///
    /// When the sink is busy (this emit happened inside a sink handler), the
    /// signal stays queued and the outer drain loop delivers it after the
    /// handler returns. A full queue or a borrowed queue refuses the signal,
    /// counts it, and owes the sink a [`AdapterFault::SignalsLost`].
    pub(super) fn emit(&self, signal: AdapterSignal) {
        match self.queue.try_borrow_mut() {
            Ok(mut queue) if queue.len() < MAX_PENDING_SIGNALS => {
                queue.push_back(signal);
                let pending = queue.len();
                self.update(|stats| stats.peak_pending = stats.peak_pending.max(pending));
            }
            Ok(_) => {
                self.update(|stats| stats.overflowed = stats.overflowed.saturating_add(1));
                let (overflowed, unqueued) = self.owed.get();
                self.owed.set((overflowed.saturating_add(1), unqueued));
            }
            // Queue borrows are scoped to push/pop with no foreign code
            // inside, so this cannot arise on the single browser thread; if it
            // does, the signal is counted and reported, never silently gone.
            Err(_) => {
                self.update(|stats| stats.unqueued = stats.unqueued.saturating_add(1));
                let (overflowed, unqueued) = self.owed.get();
                self.owed.set((overflowed, unqueued.saturating_add(1)));
            }
        }
        self.drain();
    }

    /// Delivers queued signals until the queue is empty, then any owed loss
    /// report. Returns at once when a sink handler is already on the stack:
    /// that handler's drain loop delivers on return.
    pub(super) fn drain(&self) {
        loop {
            let Ok(mut sink) = self.sink.try_borrow_mut() else {
                return;
            };
            let next = if let Ok(mut queue) = self.queue.try_borrow_mut() {
                queue.pop_front()
            } else {
                // Left queued for the drain that holds the borrow.
                self.update(|stats| {
                    stats.drain_conflicts = stats.drain_conflicts.saturating_add(1);
                });
                return;
            };
            let signal = match next {
                Some(signal) => signal,
                None => match self.owed.replace((0, 0)) {
                    (0, 0) => return,
                    (overflowed, unqueued) => AdapterSignal::Fault(AdapterFault::SignalsLost {
                        overflowed,
                        unqueued,
                    }),
                },
            };
            sink(signal);
        }
    }
}