prns_interfaces_embassy/usb_auto/

mod.rs

mod device;

pub use device::{
    WebUsbAutoClass, WebUsbAutoError, WebUsbAutoRx, WebUsbAutoState, WebUsbAutoTx,
    WEBUSB_AUTO_PACKET_SIZE,
};

use embassy_futures::select::{select, select3, Either, Either3};
use embassy_time::{with_timeout, Duration, Instant, Timer};
use embedded_io_async::{Error, ErrorKind, Read, Write};

use prns_core::interfaces::usb_auto::{
    self as contract, Capabilities, InboundReaction, Message, NodeTag,
};
use prns_core::interfaces::{
    ConnectionState, InterfaceDescriptor, InterfaceId, InterfaceKind, InterfaceStatus,
};
use prns_runtime::manifold::driver::EmbassyInterfaceStatus;
use prns_runtime::manifold::interface_seam::{
    Interface, InterfaceSeam, OutboundDisposition, OutboundDropReason,
};

const WRITE_TIMEOUT: Duration = Duration::from_millis(200);
const IO_RETRY_DELAY: Duration = Duration::from_millis(100);
const PRESENCE_PROBE_INTERVAL: Duration = Duration::from_secs(2);
const PRESENCE_STRIKES_TO_DORMANT: u8 = 2;

#[derive(Debug, PartialEq, Eq)]
enum UsbLifecycle {
    AwaitingHost,
    Linked,
    Degraded,
    Failed,
}

impl UsbLifecycle {
    fn is_linked(&self) -> bool {
        match self {
            Self::Linked | Self::Degraded => true,
            Self::AwaitingHost | Self::Failed => false,
        }
    }

    fn publish(&self, status: &EmbassyInterfaceStatus) {
        let connection = match self {
            Self::AwaitingHost => ConnectionState::Disconnected,
            Self::Linked => ConnectionState::Connected,
            Self::Degraded => ConnectionState::Degraded,
            Self::Failed => ConnectionState::Failed,
        };
        status.set_connection(connection);
    }

    fn connect(&mut self, status: &EmbassyInterfaceStatus) {
        *self = Self::Linked;
        self.publish(status);
    }

    fn recover(&mut self, status: &EmbassyInterfaceStatus) {
        if matches!(self, Self::Degraded) {
            self.connect(status);
        }
    }

    fn degrade(&mut self, status: &EmbassyInterfaceStatus) {
        match self {
            Self::Linked | Self::Degraded => {
                *self = Self::Degraded;
                self.publish(status);
            }
            Self::AwaitingHost | Self::Failed => {}
        }
    }

    fn disconnect(&mut self, status: &EmbassyInterfaceStatus) {
        if matches!(self, Self::Failed) {
            return;
        }
        *self = Self::AwaitingHost;
        self.publish(status);
    }

    fn fail(&mut self, status: &EmbassyInterfaceStatus) {
        *self = Self::Failed;
        self.publish(status);
    }

    fn disable(&mut self) {
        if !matches!(self, Self::Failed) {
            *self = Self::AwaitingHost;
        }
    }
}

#[derive(Debug, PartialEq, Eq)]
enum ReadOutcome {
    Bytes(usize),
    RetryReady,
    EndOfStream,
    TransientFailure,
    Disconnected,
    Failed,
}

#[derive(Debug, PartialEq, Eq)]
enum WriteOutcome {
    Sent(usize),
    TimedOut,
    Disabled,
    Disconnected,
    TransientFailure,
    Failed,
    Rejected,
}

enum IoEvent<'a> {
    Read(ReadOutcome),
    Outbound(&'a [u8]),
}

enum IoPriority {
    Read,
    Outbound,
}

impl IoPriority {
    fn alternate(&mut self) {
        *self = match self {
            Self::Read => Self::Outbound,
            Self::Outbound => Self::Read,
        };
    }
}

#[derive(Debug, PartialEq, Eq)]
enum PresenceVerdict {
    Present,
    SuspectedAbsent,
    Absent,
}

pub struct UsbAutoDeviceInput<'a, R, W, P> {
    pub rx: R,
    pub tx: W,
    pub status: &'a EmbassyInterfaceStatus,
    pub host_present: P,
}

pub struct UsbAutoDevice<'a, R, W, P> {
    id: InterfaceId,
    rx: R,
    tx: W,
    node_tag: NodeTag,
    status: &'a EmbassyInterfaceStatus,
    host_present: P,
}

impl<'a, R, W, P> UsbAutoDevice<'a, R, W, P> {
    #[must_use]
    pub fn new(input: UsbAutoDeviceInput<'a, R, W, P>) -> Self {
        let UsbAutoDeviceInput {
            rx,
            tx,
            status,
            host_present,
        } = input;
        let id = status.id();
        Self {
            id,
            rx,
            tx,
            node_tag: contract::node_tag_for(id),
            status,
            host_present,
        }
    }
}

impl<R, W, P> Interface for UsbAutoDevice<'_, R, W, P>
where
    R: Read,
    W: Write,
    P: FnMut() -> bool,
{
    const HW_MTU: usize = prns_core::interfaces::usb_auto::DEVICE_USB_HW_MTU;
    const KIND: InterfaceKind = InterfaceKind::UsbAutoDevice;

    fn descriptor(&self) -> InterfaceDescriptor {
        contract::device_descriptor(self.id)
    }

    fn channel_tag(&self) -> &[u8] {
        self.id.as_bytes()
    }

    async fn run<Seam: InterfaceSeam>(self, mut seam: Seam) {
        let UsbAutoDevice {
            id: _,
            mut rx,
            mut tx,
            node_tag,
            status,
            mut host_present,
        } = self;
        let mut decoder = contract::Decoder::new();
        let mut read_buf = [0u8; contract::READ_CHUNK_BYTES];
        let mut frame_buf = [0u8; contract::MAX_FRAMED_BYTES];
        let mut lifecycle = UsbLifecycle::AwaitingHost;
        let mut absent_probes = 0u8;
        let mut read_retry_at = None;
        let mut presence_probe_at = Instant::now() + PRESENCE_PROBE_INTERVAL;
        let mut io_priority = IoPriority::Read;

        lifecycle.publish(status);

        loop {
            if !status.is_enabled() {
                lifecycle.disable();
                decoder = contract::Decoder::new();
                read_retry_at = None;
                absent_probes = 0;
                status.wait_until_enabled().await;
                lifecycle.publish(status);
                presence_probe_at = Instant::now() + PRESENCE_PROBE_INTERVAL;
            }
            match select3(
                status.wait_until_disabled(),
                Timer::at(presence_probe_at),
                next_io(
                    &mut rx,
                    &mut read_buf,
                    read_retry_at,
                    &mut seam,
                    &io_priority,
                ),
            )
            .await
            {
                Either3::First(()) => {}
                Either3::Second(()) => {
                    presence_probe_at = Instant::now() + PRESENCE_PROBE_INTERVAL;
                    match presence_verdict(host_present(), &mut absent_probes) {
                        PresenceVerdict::Present | PresenceVerdict::SuspectedAbsent => {}
                        PresenceVerdict::Absent => {
                            decoder = contract::Decoder::new();
                            lifecycle.disconnect(status);
                        }
                    }
                }
                Either3::Third(event) => {
                    io_priority.alternate();
                    match event {
                        IoEvent::Read(ReadOutcome::Bytes(n)) => {
                            read_retry_at = None;
                            absent_probes = 0;
                            status.add_rx(n as u64);
                            lifecycle.recover(status);
                            for &byte in &read_buf[..n] {
                                let Ok(Some(frame)) = decoder.feed(byte) else {
                                    continue;
                                };
                                if frame.is_empty() {
                                    continue;
                                }
                                match contract::react_to(contract::decode_message(frame)) {
                                    InboundReaction::AnswerHandshake => {
                                        let ack = Message::HelloAck {
                                            tag: node_tag,
                                            capabilities: Capabilities::none(),
                                        };
                                        match write_message(&mut tx, &ack, &mut frame_buf, status)
                                            .await
                                        {
                                            WriteOutcome::Sent(n) => {
                                                status.add_tx(n as u64);
                                                lifecycle.connect(status);
                                            }
                                            WriteOutcome::TimedOut
                                            | WriteOutcome::TransientFailure => {
                                                lifecycle.degrade(status);
                                            }
                                            WriteOutcome::Disconnected => {
                                                decoder = contract::Decoder::new();
                                                lifecycle.disconnect(status);
                                            }
                                            WriteOutcome::Failed | WriteOutcome::Rejected => {
                                                decoder = contract::Decoder::new();
                                                lifecycle.fail(status);
                                            }
                                            WriteOutcome::Disabled => {}
                                        }
                                    }
                                    InboundReaction::Deliver(packet) => {
                                        if lifecycle.is_linked() && !packet.is_empty() {
                                            seam.next_inbound(packet).await;
                                        }
                                    }
                                    InboundReaction::Ignore => {}
                                }
                            }
                        }
                        IoEvent::Read(ReadOutcome::RetryReady) => {
                            read_retry_at = None;
                        }
                        IoEvent::Read(ReadOutcome::EndOfStream | ReadOutcome::Disconnected) => {
                            decoder = contract::Decoder::new();
                            lifecycle.disconnect(status);
                            read_retry_at = Some(Instant::now() + IO_RETRY_DELAY);
                        }
                        IoEvent::Read(ReadOutcome::TransientFailure) => {
                            lifecycle.degrade(status);
                            read_retry_at = Some(Instant::now() + IO_RETRY_DELAY);
                        }
                        IoEvent::Read(ReadOutcome::Failed) => {
                            decoder = contract::Decoder::new();
                            lifecycle.fail(status);
                            read_retry_at = Some(Instant::now() + IO_RETRY_DELAY);
                        }
                        IoEvent::Outbound(out) => {
                            let disposition = if !lifecycle.is_linked() {
                                OutboundDisposition::Dropped(OutboundDropReason::Disconnected)
                            } else {
                                let data = Message::Data(out);
                                match write_message(&mut tx, &data, &mut frame_buf, status).await {
                                    WriteOutcome::Sent(n) => {
                                        status.add_tx(n as u64);
                                        lifecycle.recover(status);
                                        OutboundDisposition::Sent
                                    }
                                    WriteOutcome::TimedOut => {
                                        lifecycle.degrade(status);
                                        OutboundDisposition::Dropped(OutboundDropReason::TimedOut)
                                    }
                                    WriteOutcome::Disabled => {
                                        OutboundDisposition::Dropped(OutboundDropReason::Disabled)
                                    }
                                    WriteOutcome::Disconnected => {
                                        decoder = contract::Decoder::new();
                                        lifecycle.disconnect(status);
                                        OutboundDisposition::Dropped(
                                            OutboundDropReason::Disconnected,
                                        )
                                    }
                                    WriteOutcome::TransientFailure => {
                                        lifecycle.degrade(status);
                                        OutboundDisposition::Dropped(
                                            OutboundDropReason::TransportFailure,
                                        )
                                    }
                                    WriteOutcome::Failed => {
                                        lifecycle.fail(status);
                                        OutboundDisposition::Dropped(
                                            OutboundDropReason::TransportFailure,
                                        )
                                    }
                                    WriteOutcome::Rejected => {
                                        lifecycle.fail(status);
                                        OutboundDisposition::Dropped(OutboundDropReason::Rejected)
                                    }
                                }
                            };
                            seam.complete_outbound(disposition);
                        }
                    }
                }
            }
        }
    }
}

async fn next_io<'a, R, Seam>(
    rx: &'a mut R,
    read_buf: &'a mut [u8; contract::READ_CHUNK_BYTES],
    read_retry_at: Option<Instant>,
    seam: &'a mut Seam,
    priority: &IoPriority,
) -> IoEvent<'a>
where
    R: Read,
    Seam: InterfaceSeam,
{
    if matches!(priority, IoPriority::Outbound) {
        return match select(seam.next_outbound(), read_once(rx, read_buf, read_retry_at)).await {
            Either::First(out) => IoEvent::Outbound(out),
            Either::Second(outcome) => IoEvent::Read(outcome),
        };
    }
    match select(read_once(rx, read_buf, read_retry_at), seam.next_outbound()).await {
        Either::First(outcome) => IoEvent::Read(outcome),
        Either::Second(out) => IoEvent::Outbound(out),
    }
}

async fn read_once<R: Read>(
    rx: &mut R,
    read_buf: &mut [u8; contract::READ_CHUNK_BYTES],
    retry_at: Option<Instant>,
) -> ReadOutcome {
    if let Some(retry_at) = retry_at {
        Timer::at(retry_at).await;
        return ReadOutcome::RetryReady;
    }
    match rx.read(read_buf).await {
        Ok(0) => ReadOutcome::EndOfStream,
        Ok(n) => ReadOutcome::Bytes(n),
        Err(error) => match classify_io_error(error.kind()) {
            IoFailure::Transient => ReadOutcome::TransientFailure,
            IoFailure::Disconnected => ReadOutcome::Disconnected,
            IoFailure::Failed => ReadOutcome::Failed,
        },
    }
}

#[derive(Debug, PartialEq, Eq)]
enum IoFailure {
    Transient,
    Disconnected,
    Failed,
}

fn classify_io_error(kind: ErrorKind) -> IoFailure {
    match kind {
        ErrorKind::TimedOut | ErrorKind::Interrupted => IoFailure::Transient,
        ErrorKind::NotFound
        | ErrorKind::ConnectionRefused
        | ErrorKind::ConnectionReset
        | ErrorKind::ConnectionAborted
        | ErrorKind::NotConnected
        | ErrorKind::AddrNotAvailable
        | ErrorKind::BrokenPipe => IoFailure::Disconnected,
        _ => IoFailure::Failed,
    }
}

fn presence_verdict(present: bool, absent_probes: &mut u8) -> PresenceVerdict {
    if present {
        *absent_probes = 0;
        return PresenceVerdict::Present;
    }
    *absent_probes = absent_probes.saturating_add(1);
    if *absent_probes >= PRESENCE_STRIKES_TO_DORMANT {
        PresenceVerdict::Absent
    } else {
        PresenceVerdict::SuspectedAbsent
    }
}

async fn write_message<W: Write>(
    tx: &mut W,
    message: &Message<'_>,
    frame_buf: &mut [u8; contract::MAX_FRAMED_BYTES],
    status: &EmbassyInterfaceStatus,
) -> WriteOutcome {
    let Ok(n) = message.write_framed(frame_buf) else {
        return WriteOutcome::Rejected;
    };
    match select(
        status.wait_until_disabled(),
        with_timeout(WRITE_TIMEOUT, tx.write_all(&frame_buf[..n])),
    )
    .await
    {
        Either::First(()) => WriteOutcome::Disabled,
        Either::Second(Err(_)) => WriteOutcome::TimedOut,
        Either::Second(Ok(Ok(()))) => WriteOutcome::Sent(n),
        Either::Second(Ok(Err(error))) => match classify_io_error(error.kind()) {
            IoFailure::Transient => WriteOutcome::TransientFailure,
            IoFailure::Disconnected => WriteOutcome::Disconnected,
            IoFailure::Failed => WriteOutcome::Failed,
        },
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use prns_core::interfaces::{FrameSink, InterfaceOriginKind, InterfaceStatus, IFAC_MAX_SIZE};
    use prns_runtime::manifold::driver::{leaked_grant_lane, EmbassyInterfaceSeam};
    use prns_runtime::manifold::grant::{GrantConsumer, GrantProducer};

    use ::core::cell::{Cell, RefCell};
    use ::core::convert::Infallible;
    use ::core::future::pending;
    use embassy_futures::block_on;
    use embassy_futures::join::join;
    use embassy_futures::select::{select, Either};
    use embassy_futures::yield_now;
    use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
    use embassy_sync::channel::Channel;
    use embassy_time::{with_timeout, Duration};
    use std::collections::VecDeque;

    const WATCHDOG: Duration = Duration::from_secs(5);

    const DEVICE_SLOT: usize = prns_core::interfaces::usb_auto::DEVICE_USB_HW_MTU + IFAC_MAX_SIZE;

    struct MockStream<'a> {
        buf: &'a RefCell<VecDeque<u8>>,
    }

    struct RecordingSeam<'a, S> {
        inner: S,
        dispositions: &'a RefCell<Vec<OutboundDisposition>>,
    }

    impl<S: InterfaceSeam> InterfaceSeam for RecordingSeam<'_, S> {
        fn interface_origin(&self) -> InterfaceOriginKind {
            self.inner.interface_origin()
        }

        fn fill_entropy(&mut self, bytes: &mut [u8]) {
            self.inner.fill_entropy(bytes);
        }

        async fn inbound_sink(&mut self) -> &mut dyn FrameSink {
            self.inner.inbound_sink().await
        }

        async fn commit_inbound(&mut self) {
            self.inner.commit_inbound().await;
        }

        async fn next_inbound(&mut self, frame: &[u8]) {
            self.inner.next_inbound(frame).await;
        }

        async fn next_outbound(&mut self) -> &[u8] {
            self.inner.next_outbound().await
        }

        fn complete_outbound(&mut self, disposition: OutboundDisposition) {
            self.inner.complete_outbound(disposition.clone());
            self.dispositions.borrow_mut().push(disposition);
        }
    }

    #[derive(Debug)]
    struct MockIoError(ErrorKind);

    impl embedded_io_async::Error for MockIoError {
        fn kind(&self) -> ErrorKind {
            self.0
        }
    }

    enum MockReadAction {
        Bytes(Vec<u8>),
        EndOfStream,
        Error(ErrorKind),
        Pending,
    }

    struct ScriptedReader<'a> {
        actions: &'a RefCell<VecDeque<MockReadAction>>,
        calls: &'a Cell<usize>,
        cancellations: &'a Cell<usize>,
    }

    struct CancellationGuard<'a> {
        cancellations: &'a Cell<usize>,
    }

    impl Drop for CancellationGuard<'_> {
        fn drop(&mut self) {
            self.cancellations.set(self.cancellations.get() + 1);
        }
    }

    impl embedded_io_async::ErrorType for ScriptedReader<'_> {
        type Error = MockIoError;
    }

    impl Read for ScriptedReader<'_> {
        async fn read(&mut self, out: &mut [u8]) -> Result<usize, Self::Error> {
            self.calls.set(self.calls.get() + 1);
            let action = self.actions.borrow_mut().pop_front();
            match action {
                Some(MockReadAction::Bytes(bytes)) => {
                    let n = bytes.len().min(out.len());
                    out[..n].copy_from_slice(&bytes[..n]);
                    Ok(n)
                }
                Some(MockReadAction::EndOfStream) => Ok(0),
                Some(MockReadAction::Error(kind)) => Err(MockIoError(kind)),
                Some(MockReadAction::Pending) | None => {
                    let _guard = CancellationGuard {
                        cancellations: self.cancellations,
                    };
                    pending().await
                }
            }
        }
    }

    enum MockWriteAction {
        Accept,
        Error(ErrorKind),
        Pending,
    }

    struct ScriptedWriter<'a> {
        actions: &'a RefCell<VecDeque<MockWriteAction>>,
        cancellations: &'a Cell<usize>,
    }

    impl embedded_io_async::ErrorType for ScriptedWriter<'_> {
        type Error = MockIoError;
    }

    impl Write for ScriptedWriter<'_> {
        async fn write(&mut self, data: &[u8]) -> Result<usize, Self::Error> {
            let action = self.actions.borrow_mut().pop_front();
            match action {
                Some(MockWriteAction::Accept) => Ok(data.len()),
                Some(MockWriteAction::Error(kind)) => Err(MockIoError(kind)),
                Some(MockWriteAction::Pending) | None => {
                    let _guard = CancellationGuard {
                        cancellations: self.cancellations,
                    };
                    pending().await
                }
            }
        }

        async fn flush(&mut self) -> Result<(), Self::Error> {
            Ok(())
        }
    }

    impl embedded_io_async::ErrorType for MockStream<'_> {
        type Error = Infallible;
    }

    impl Read for MockStream<'_> {
        async fn read(&mut self, out: &mut [u8]) -> Result<usize, Self::Error> {
            loop {
                {
                    let mut queue = self.buf.borrow_mut();
                    if !queue.is_empty() {
                        let n = queue.len().min(out.len());
                        for slot in out.iter_mut().take(n) {
                            *slot = queue.pop_front().expect("non-empty");
                        }
                        return Ok(n);
                    }
                }
                yield_now().await;
            }
        }
    }

    impl Write for MockStream<'_> {
        async fn write(&mut self, data: &[u8]) -> Result<usize, Self::Error> {
            self.buf.borrow_mut().extend(data.iter().copied());
            Ok(data.len())
        }
    }

    fn device_id() -> InterfaceId {
        InterfaceId::new([0xD0; 8])
    }

    async fn read_until<T>(
        wire: &RefCell<VecDeque<u8>>,
        decoder: &mut contract::Decoder,
        mut pick: impl FnMut(Message<'_>) -> Option<T>,
    ) -> T {
        loop {
            let byte = loop {
                if let Some(byte) = wire.borrow_mut().pop_front() {
                    break byte;
                }
                yield_now().await;
            };
            if let Ok(Some(frame)) = decoder.feed(byte) {
                if !frame.is_empty() {
                    if let Ok(message) = contract::decode_message(frame) {
                        if let Some(picked) = pick(message) {
                            return picked;
                        }
                    }
                }
            }
        }
    }

    #[test]
    fn the_device_handshakes_a_host_then_carries_data_both_ways() {
        let host_to_device = RefCell::new(VecDeque::new());
        let device_to_host = RefCell::new(VecDeque::new());
        let dispositions = RefCell::new(Vec::new());
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Initializing);

        let notify: Channel<CriticalSectionRawMutex, InterfaceId, 2> = Channel::new();
        let (in_tx, mut in_rx) = leaked_grant_lane::<DEVICE_SLOT>(2);
        let (mut out_tx, out_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);

        block_on(async {
            let device = UsbAutoDevice::new(UsbAutoDeviceInput {
                rx: MockStream {
                    buf: &host_to_device,
                },
                tx: MockStream {
                    buf: &device_to_host,
                },
                status: &status,
                host_present: || true,
            });
            let inner =
                EmbassyInterfaceSeam::new(device_id(), in_tx, notify.sender(), out_rx, |bytes| {
                    bytes.fill(0)
                });
            let seam = RecordingSeam {
                inner,
                dispositions: &dispositions,
            };
            let device_run = device.run(seam);

            let driver = async {
                let mut frame = [0u8; contract::MAX_FRAMED_BYTES];
                let mut decoder = contract::Decoder::new();

                let hello = Message::Hello(Capabilities::host());
                let n = hello.write_framed(&mut frame).expect("frames the hello");
                host_to_device
                    .borrow_mut()
                    .extend(frame[..n].iter().copied());

                read_until(&device_to_host, &mut decoder, |message| {
                    matches!(message, Message::HelloAck { .. }).then_some(())
                })
                .await;
                assert_eq!(status.connection(), ConnectionState::Connected);

                let inbound_packet = [0xAAu8, 0xBB, 0xCC, 0xDD];
                let data = Message::Data(&inbound_packet);
                let n = data.write_framed(&mut frame).expect("frames the data");
                host_to_device
                    .borrow_mut()
                    .extend(frame[..n].iter().copied());
                assert_eq!(notify.receive().await, device_id());
                let received = in_rx.peek().await;
                assert_eq!(received.frame(), &inbound_packet);
                in_rx.release();

                let outbound_packet = [0x11u8, 0x22, 0x33];
                out_tx.grant().await.fill_for(device_id(), &outbound_packet);
                out_tx.commit();
                let delivered =
                    read_until(&device_to_host, &mut decoder, |message| match message {
                        Message::Data(packet) => Some(packet.to_vec()),
                        _ => None,
                    })
                    .await;
                assert_eq!(delivered, outbound_packet);
                while dispositions.borrow().is_empty() {
                    yield_now().await;
                }
                assert_eq!(
                    dispositions.borrow().as_slice(),
                    &[OutboundDisposition::Sent]
                );
                let next = with_timeout(WATCHDOG, out_tx.grant())
                    .await
                    .expect("completed outbound releases its lane slot");
                next.fill_for(device_id(), &[0x44]);
            };

            match select(device_run, with_timeout(WATCHDOG, driver)).await {
                Either::Second(result) => result.expect("the link completes before the watchdog"),
                Either::First(()) => unreachable!("the device loop never returns"),
            }
        });
    }

    #[test]
    fn outbound_while_unlinked_is_typed_and_releases_its_lane_slot() {
        let host_to_device = RefCell::new(VecDeque::new());
        let device_to_host = RefCell::new(VecDeque::new());
        let dispositions = RefCell::new(Vec::new());
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Initializing);
        let notify: Channel<CriticalSectionRawMutex, InterfaceId, 1> = Channel::new();
        let (in_tx, _in_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);
        let (mut out_tx, out_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);

        block_on(async {
            let device = UsbAutoDevice::new(UsbAutoDeviceInput {
                rx: MockStream {
                    buf: &host_to_device,
                },
                tx: MockStream {
                    buf: &device_to_host,
                },
                status: &status,
                host_present: || true,
            });
            let inner =
                EmbassyInterfaceSeam::new(device_id(), in_tx, notify.sender(), out_rx, |bytes| {
                    bytes.fill(0)
                });
            let seam = RecordingSeam {
                inner,
                dispositions: &dispositions,
            };
            let device_run = device.run(seam);

            let driver = async {
                out_tx.grant().await.fill_for(device_id(), &[0x55]);
                out_tx.commit();
                while dispositions.borrow().is_empty() {
                    yield_now().await;
                }
                assert_eq!(
                    dispositions.borrow().as_slice(),
                    &[OutboundDisposition::Dropped(
                        OutboundDropReason::Disconnected
                    )]
                );
                with_timeout(WATCHDOG, out_tx.grant())
                    .await
                    .expect("discarded outbound releases its lane slot");
            };

            match select(device_run, with_timeout(WATCHDOG, driver)).await {
                Either::Second(result) => {
                    result.expect("the discard completes before the watchdog")
                }
                Either::First(()) => unreachable!("the device loop never returns"),
            }
        });
    }

    #[test]
    fn read_outcomes_distinguish_bytes_eof_transient_disconnect_and_failure() {
        let actions = RefCell::new(VecDeque::from([
            MockReadAction::Bytes(vec![0xAA, 0xBB]),
            MockReadAction::EndOfStream,
            MockReadAction::Error(ErrorKind::Interrupted),
            MockReadAction::Error(ErrorKind::NotConnected),
            MockReadAction::Error(ErrorKind::Other),
        ]));
        let calls = Cell::new(0);
        let cancellations = Cell::new(0);
        let mut reader = ScriptedReader {
            actions: &actions,
            calls: &calls,
            cancellations: &cancellations,
        };
        let mut read_buf = [0u8; contract::READ_CHUNK_BYTES];

        block_on(async {
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::Bytes(2)
            );
            assert_eq!(&read_buf[..2], &[0xAA, 0xBB]);
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::EndOfStream
            );
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::TransientFailure
            );
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::Disconnected
            );
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::Failed
            );
        });
        assert_eq!(calls.get(), 5);
        assert_eq!(cancellations.get(), 0);
    }

    #[test]
    fn retry_backoff_does_not_poll_an_immediately_failing_reader() {
        let actions = RefCell::new(VecDeque::from([MockReadAction::Error(ErrorKind::Other)]));
        let calls = Cell::new(0);
        let cancellations = Cell::new(0);
        let mut reader = ScriptedReader {
            actions: &actions,
            calls: &calls,
            cancellations: &cancellations,
        };
        let mut read_buf = [0u8; contract::READ_CHUNK_BYTES];

        block_on(async {
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::Failed
            );
            assert_eq!(
                read_once(&mut reader, &mut read_buf, Some(Instant::now())).await,
                ReadOutcome::RetryReady
            );
        });
        assert_eq!(calls.get(), 1);
    }

    #[test]
    fn disabling_cancels_a_blocked_write_before_its_timeout() {
        let actions = RefCell::new(VecDeque::from([MockWriteAction::Pending]));
        let cancellations = Cell::new(0);
        let mut writer = ScriptedWriter {
            actions: &actions,
            cancellations: &cancellations,
        };
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Connected);
        let mut frame_buf = [0u8; contract::MAX_FRAMED_BYTES];

        block_on(async {
            let (outcome, ()) = join(
                write_message(
                    &mut writer,
                    &Message::Data(&[0x11]),
                    &mut frame_buf,
                    &status,
                ),
                async {
                    yield_now().await;
                    status.disable();
                },
            )
            .await;
            assert_eq!(outcome, WriteOutcome::Disabled);
        });
        assert_eq!(cancellations.get(), 1);
    }

    #[test]
    fn write_outcomes_preserve_transport_failure_structure() {
        let actions = RefCell::new(VecDeque::from([
            MockWriteAction::Accept,
            MockWriteAction::Error(ErrorKind::NotConnected),
            MockWriteAction::Error(ErrorKind::Interrupted),
            MockWriteAction::Error(ErrorKind::Other),
        ]));
        let cancellations = Cell::new(0);
        let mut writer = ScriptedWriter {
            actions: &actions,
            cancellations: &cancellations,
        };
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Connected);
        let mut frame_buf = [0u8; contract::MAX_FRAMED_BYTES];

        block_on(async {
            assert!(matches!(
                write_message(
                    &mut writer,
                    &Message::Data(&[0x11]),
                    &mut frame_buf,
                    &status
                )
                .await,
                WriteOutcome::Sent(_)
            ));
            assert_eq!(
                write_message(
                    &mut writer,
                    &Message::Data(&[0x22]),
                    &mut frame_buf,
                    &status
                )
                .await,
                WriteOutcome::Disconnected
            );
            assert_eq!(
                write_message(
                    &mut writer,
                    &Message::Data(&[0x33]),
                    &mut frame_buf,
                    &status
                )
                .await,
                WriteOutcome::TransientFailure
            );
            assert_eq!(
                write_message(
                    &mut writer,
                    &Message::Data(&[0x44]),
                    &mut frame_buf,
                    &status
                )
                .await,
                WriteOutcome::Failed
            );
        });
    }

    #[test]
    fn blocked_write_times_out_and_cancels_the_transport_future() {
        let actions = RefCell::new(VecDeque::from([MockWriteAction::Pending]));
        let cancellations = Cell::new(0);
        let mut writer = ScriptedWriter {
            actions: &actions,
            cancellations: &cancellations,
        };
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Connected);
        let mut frame_buf = [0u8; contract::MAX_FRAMED_BYTES];

        block_on(async {
            assert_eq!(
                write_message(
                    &mut writer,
                    &Message::Data(&[0x77]),
                    &mut frame_buf,
                    &status
                )
                .await,
                WriteOutcome::TimedOut
            );
        });
        assert_eq!(cancellations.get(), 1);
    }

    #[test]
    fn failure_state_survives_disable_and_reenable() {
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Initializing);
        let mut lifecycle = UsbLifecycle::Failed;
        lifecycle.publish(&status);
        assert_eq!(status.connection(), ConnectionState::Failed);

        status.disable();
        lifecycle.disable();
        assert_eq!(status.connection(), ConnectionState::Disabled);

        status.enable();
        lifecycle.publish(&status);
        assert_eq!(status.connection(), ConnectionState::Failed);
    }

    #[test]
    fn data_before_handshake_is_not_delivered() {
        let host_to_device = RefCell::new(VecDeque::new());
        let device_to_host = RefCell::new(VecDeque::new());
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Initializing);
        let notify: Channel<CriticalSectionRawMutex, InterfaceId, 1> = Channel::new();
        let (in_tx, _in_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);
        let (_out_tx, out_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);
        let mut frame = [0u8; contract::MAX_FRAMED_BYTES];
        let n = Message::Data(&[0xAA, 0xBB])
            .write_framed(&mut frame)
            .expect("frames pre-handshake data");
        host_to_device
            .borrow_mut()
            .extend(frame[..n].iter().copied());

        block_on(async {
            let device = UsbAutoDevice::new(UsbAutoDeviceInput {
                rx: MockStream {
                    buf: &host_to_device,
                },
                tx: MockStream {
                    buf: &device_to_host,
                },
                status: &status,
                host_present: || true,
            });
            let seam =
                EmbassyInterfaceSeam::new(device_id(), in_tx, notify.sender(), out_rx, |bytes| {
                    bytes.fill(0)
                });
            match select(
                device.run(seam),
                with_timeout(Duration::from_millis(20), notify.receive()),
            )
            .await
            {
                Either::Second(result) => {
                    assert!(result.is_err(), "pre-handshake data reached the manifold");
                }
                Either::First(()) => unreachable!("the device loop never returns"),
            }
        });
        assert_eq!(status.connection(), ConnectionState::Disconnected);
    }

    #[test]
    fn failed_hello_ack_never_publishes_connected() {
        let mut frame = [0u8; contract::MAX_FRAMED_BYTES];
        let n = Message::Hello(Capabilities::host())
            .write_framed(&mut frame)
            .expect("frames hello");
        let read_actions = RefCell::new(VecDeque::from([
            MockReadAction::Bytes(frame[..n].to_vec()),
            MockReadAction::Pending,
        ]));
        let read_calls = Cell::new(0);
        let read_cancellations = Cell::new(0);
        let write_actions = RefCell::new(VecDeque::from([MockWriteAction::Error(
            ErrorKind::NotConnected,
        )]));
        let write_cancellations = Cell::new(0);
        let status = EmbassyInterfaceStatus::new(device_id(), ConnectionState::Initializing);
        let notify: Channel<CriticalSectionRawMutex, InterfaceId, 1> = Channel::new();
        let (in_tx, _in_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);
        let (_out_tx, out_rx) = leaked_grant_lane::<DEVICE_SLOT>(1);

        block_on(async {
            let device = UsbAutoDevice::new(UsbAutoDeviceInput {
                rx: ScriptedReader {
                    actions: &read_actions,
                    calls: &read_calls,
                    cancellations: &read_cancellations,
                },
                tx: ScriptedWriter {
                    actions: &write_actions,
                    cancellations: &write_cancellations,
                },
                status: &status,
                host_present: || true,
            });
            let seam =
                EmbassyInterfaceSeam::new(device_id(), in_tx, notify.sender(), out_rx, |bytes| {
                    bytes.fill(0)
                });
            match select(device.run(seam), Timer::after(Duration::from_millis(20))).await {
                Either::Second(()) => {}
                Either::First(()) => unreachable!("the device loop never returns"),
            }
        });
        assert_eq!(status.connection(), ConnectionState::Disconnected);
        assert!(write_actions.borrow().is_empty());
    }

    #[test]
    fn cancelled_pending_read_consumes_no_scripted_data() {
        let actions = RefCell::new(VecDeque::from([
            MockReadAction::Pending,
            MockReadAction::Bytes(vec![0xAB]),
        ]));
        let calls = Cell::new(0);
        let cancellations = Cell::new(0);
        let mut reader = ScriptedReader {
            actions: &actions,
            calls: &calls,
            cancellations: &cancellations,
        };
        let mut read_buf = [0u8; contract::READ_CHUNK_BYTES];

        block_on(async {
            assert!(matches!(
                select(read_once(&mut reader, &mut read_buf, None), async {
                    yield_now().await
                })
                .await,
                Either::Second(())
            ));
            assert_eq!(cancellations.get(), 1);
            assert_eq!(
                read_once(&mut reader, &mut read_buf, None).await,
                ReadOutcome::Bytes(1)
            );
            assert_eq!(read_buf[0], 0xAB);
        });
    }

    #[test]
    fn presence_present_clears_strikes_and_holds_the_link() {
        let mut absent = 0u8;
        assert_eq!(
            presence_verdict(true, &mut absent),
            PresenceVerdict::Present
        );
        assert_eq!(absent, 0);

        absent = 1;
        assert_eq!(
            presence_verdict(true, &mut absent),
            PresenceVerdict::Present
        );
        assert_eq!(absent, 0);
    }

    #[test]
    fn presence_absent_disconnects_only_after_the_strike_threshold() {
        let mut absent = 0u8;
        assert_eq!(
            presence_verdict(false, &mut absent),
            PresenceVerdict::SuspectedAbsent
        );
        assert_eq!(absent, 1);
        assert_eq!(
            presence_verdict(false, &mut absent),
            PresenceVerdict::Absent
        );

        let mut recovered = 1u8;
        assert_eq!(
            presence_verdict(true, &mut recovered),
            PresenceVerdict::Present
        );
        assert_eq!(
            presence_verdict(false, &mut recovered),
            PresenceVerdict::SuspectedAbsent
        );
    }
}