liminal-sdk 0.19.1

//! The signal port's bound and loss accounting, pinned natively.

use alloc::boxed::Box;
use alloc::rc::{Rc, Weak};
use alloc::vec::Vec;
use core::cell::RefCell;

use super::super::super::core::DriverOutput;
use super::{AdapterFault, AdapterSignal, MAX_PENDING_SIGNALS, SignalPort};

type Received = Rc<RefCell<Vec<AdapterSignal>>>;

/// A port whose sink records every signal and, on the first one, re-enters
/// the port `nested` times before returning, the way a sink handler calling
/// back into the adapter does.
fn reentrant_port(nested: usize) -> (Rc<SignalPort>, Received) {
    let received: Received = Rc::new(RefCell::new(Vec::new()));
    let slot: Rc<RefCell<Option<Weak<SignalPort>>>> = Rc::new(RefCell::new(None));
    let sink_received = Rc::clone(&received);
    let sink_slot = Rc::clone(&slot);
    let port = Rc::new(SignalPort::new(Box::new(move |signal| {
        let first = sink_received.borrow().is_empty();
        sink_received.borrow_mut().push(signal);
        if first {
            let port = sink_slot.borrow().as_ref().and_then(Weak::upgrade);
            if let Some(port) = port {
                for _ in 0..nested {
                    port.emit(AdapterSignal::Output(DriverOutput::Opened));
                }
            }
        }
    })));
    // A weak handle: the sink must not keep its own port alive.
    *slot.borrow_mut() = Some(Rc::downgrade(&port));
    (port, received)
}

#[test]
fn a_reentrant_burst_past_the_bound_is_capped_and_reported_by_name() {
    let excess = 5_u64;
    let nested = MAX_PENDING_SIGNALS + 5;
    let (port, received) = reentrant_port(nested);

    port.emit(AdapterSignal::Output(DriverOutput::Opened));

    let stats = port.stats();
    assert_eq!(
        stats.peak_pending, MAX_PENDING_SIGNALS,
        "held at most the bound"
    );
    assert_eq!(stats.overflowed, excess);
    let received = received.borrow();
    assert_eq!(received.len(), 1 + MAX_PENDING_SIGNALS + 1);
    assert_eq!(
        received.last(),
        Some(&AdapterSignal::Fault(AdapterFault::SignalsLost {
            overflowed: excess,
            unqueued: 0,
        }))
    );
}

#[test]
fn a_reentrant_burst_within_the_bound_loses_nothing() {
    let (port, received) = reentrant_port(MAX_PENDING_SIGNALS);

    port.emit(AdapterSignal::Output(DriverOutput::Opened));

    assert_eq!(port.stats().overflowed, 0);
    assert_eq!(received.borrow().len(), 1 + MAX_PENDING_SIGNALS);
    assert!(
        received
            .borrow()
            .iter()
            .all(|signal| *signal == AdapterSignal::Output(DriverOutput::Opened))
    );
}

#[test]
fn a_signal_emitted_while_the_queue_is_borrowed_is_counted_and_reported() {
    let (port, received) = reentrant_port(0);

    let held = port.queue.borrow_mut();
    port.emit(AdapterSignal::Output(DriverOutput::Opened));
    drop(held);
    let stats = port.stats();
    assert_eq!(
        stats.unqueued, 1,
        "the emit that could not queue is counted"
    );
    assert_eq!(
        stats.drain_conflicts, 1,
        "the drain that could not pop is counted"
    );
    assert!(received.borrow().is_empty());

    port.drain();

    assert_eq!(
        *received.borrow(),
        [AdapterSignal::Fault(AdapterFault::SignalsLost {
            overflowed: 0,
            unqueued: 1,
        })]
    );
}

#[test]
fn a_drain_that_finds_the_queue_borrowed_leaves_the_signal_for_later() {
    let (port, received) = reentrant_port(0);
    port.queue
        .borrow_mut()
        .push_back(AdapterSignal::Output(DriverOutput::Opened));

    let held = port.queue.borrow_mut();
    port.drain();
    drop(held);
    assert_eq!(port.stats().drain_conflicts, 1);
    assert!(received.borrow().is_empty());

    port.drain();

    assert_eq!(
        *received.borrow(),
        [AdapterSignal::Output(DriverOutput::Opened)]
    );
}

#[test]
fn the_adapters_own_faults_reach_the_sink_in_order() {
    let (port, received) = reentrant_port(0);
    let faults = [
        AdapterSignal::Fault(AdapterFault::SeamReentered),
        AdapterSignal::Fault(AdapterFault::CommandExecutionFailed {
            description: "close failed".into(),
        }),
    ];

    for fault in faults.clone() {
        port.emit(fault);
    }

    assert_eq!(*received.borrow(), faults);
}