hyperlight_common/virtq/

ring.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 The Hyperlight Authors.

//! Packed Virtqueue Ring Implementation
//!
//! This module implements the packed virtqueue format from the VIRTIO specification.
//! Packed virtqueues use a single descriptor ring where descriptors cycle through
//! available and used states, providing better cache locality and simpler memory
//! layout compared to split virtqueues.
//!
//! # Descriptor State Machine
//!
//! Each descriptor transitions through states using AVAIL and USED flags:
//!
//! ```text
//!                    Driver publishes
//!     ┌─────────┐    (AVAIL=wrap)     ┌───────────┐
//!     │  Free   │ ──────────────────> │ Available │
//!     └─────────┘                     └───────────┘
//!          ^                                │
//!          │                                │ Device consumes
//!          │ Driver reclaims                │ and marks used
//!          │ (polls USED=wrap)              │ (USED=wrap)
//!          │                                v
//!     ┌─────────┐                     ┌───────────┐
//!     │Reclaimed│ <────────────────── │   Used    │
//!     └─────────┘                     └───────────┘
//! ```
//!
//! # Wrap Counter
//!
//! The wrap counter solves ring wraparound ambiguity. When cursors wrap around
//! the ring, the wrap counter toggles, changing how AVAIL/USED flags are interpreted:
//!
//! - **wrap=true**: AVAIL=1, USED=0 means "available"; AVAIL=1, USED=1 means "used"
//! - **wrap=false**: AVAIL=0, USED=1 means "available"; AVAIL=0, USED=0 means "used"
//!
//! # Buffer Chains
//!
//! Multiple buffers can be chained using the NEXT flag. All descriptors in a chain
//! share the same ID, and only the head descriptor's AVAIL/USED flags matter for
//! state transitions:
//!
//! ```text
//! Chain with 3 buffers (ID=5):
//! ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
//! │ Desc[0]      │    │ Desc[1]      │    │ Desc[2]      │
//! │ id=42        │───>│ id=42        │───>│ id=42        │
//! │ flags=NEXT   │    │ flags=NEXT   │    │ flags=0      │
//! │ AVAIL/USED   │    │ (ignored)    │    │ (ignored)    │
//! └──────────────┘    └──────────────┘    └──────────────┘
//!       HEAD              MIDDLE               TAIL
//! ```
//!
//! # Event Suppression
//!
//! Both sides can control when they want to be notified:
//!
//! - **ENABLE**: Always notify (default)
//! - **DISABLE**: Never notify (for polling mode)
//! - **DESC**: Notify only when a specific descriptor index is reached
//! ```

use core::fmt;
use core::marker::PhantomData;
use core::sync::atomic::{Ordering, fence};

use bytemuck::Zeroable;
use smallvec::SmallVec;
use thiserror::Error;

use super::desc::{DescFlags, DescTable, Descriptor};
use super::event::{EventFlags, EventSuppression};
use super::{Layout, MemOps};

/// A single buffer element in a scatter-gather list.
///
/// Represents one contiguous memory region that the device will read from
/// or write to. Multiple elements can be chained together to form a
/// [`BufferChain`].
#[derive(Debug, Copy, Clone, Zeroable)]
pub struct BufferElement {
    /// Physical address of buffer
    pub addr: u64,
    /// Length of the buffer in bytes
    pub len: u32,
    /// Whether this buffer is writable by the device
    pub writable: bool,
}

/// A buffer returned from the ring after being used by the device.
///
/// When the device completes processing a buffer chain, it returns this
/// structure containing the original descriptor ID and the number of bytes
/// written (for chains with writable buffers).
#[derive(Debug, Copy, Clone)]
pub struct UsedBuffer {
    /// Descriptor ID that was assigned when the buffer was submitted
    pub id: u16,
    /// Number of bytes written by the device to writable buffers.
    /// For read-only chains, this may be 0 or the total readable length.
    pub len: u32,
}

/// Result of submitting a buffer to the ring.
///
/// Contains the assigned descriptor ID and whether the other side
/// needs to be notified about the new buffer.
#[derive(Debug, Copy, Clone)]
pub struct SubmitResult {
    /// Descriptor ID assigned to the submitted buffer chain
    /// Use this ID to correlate completions with submissions.
    pub id: u16,
    /// Whether the device should be notified immediately based on the other
    /// side's event suppression settings.
    pub notify: bool,
}

/// Memory operation that failed in the backend.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum MemOp {
    /// Reading a descriptor from the descriptor table.
    ReadDesc,
    /// Writing a descriptor to the descriptor table.
    WriteDesc,
    /// Reading an event suppression structure.
    ReadEvent,
    /// Writing an event suppression structure.
    WriteEvent,
}

impl fmt::Display for MemOp {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::ReadDesc => f.write_str("reading descriptor"),
            Self::WriteDesc => f.write_str("writing descriptor"),
            Self::ReadEvent => f.write_str("reading event suppression"),
            Self::WriteEvent => f.write_str("writing event suppression"),
        }
    }
}

#[derive(Error, Debug)]
pub enum RingError {
    #[error("Buffer chain is empty")]
    EmptyChain,
    #[error("Buffer chain is malformed")]
    BadChain,
    #[error("Operation would block")]
    WouldBlock,
    #[error("Out of memory")]
    OutOfMemory,
    #[error("Invalid state")]
    InvalidState,
    #[error("Invalid memory layout")]
    InvalidLayout,
    #[error("Backend memory error while {op} at address 0x{addr:x}, len {len}")]
    MemError {
        /// Memory operation that failed.
        op: MemOp,
        /// Address passed to the memory backend.
        addr: u64,
        /// Number of bytes requested for the operation.
        len: usize,
    },
}

impl RingError {
    #[inline]
    fn mem_err(op: MemOp, addr: u64) -> Self {
        let len = match op {
            MemOp::ReadDesc | MemOp::WriteDesc => Descriptor::SIZE,
            MemOp::ReadEvent | MemOp::WriteEvent => EventSuppression::SIZE,
        };

        Self::MemError { op, addr, len }
    }
}

/// Type-state: Can add readable buffers
pub struct Readable;

/// Type-state: Can add writable buffers (no more readables allowed)
pub struct Writable;

/// A builder for buffer chains using type-state to enforce readable/writable order.
///
/// Upholds invariants: at least one buffer must be present in the chain,
/// and readable buffers must be added before writable buffers.
///
/// The builder stores up to 16 buffer elements inline to avoid allocation for
/// common small chains. Larger chains are still supported and spill to the heap.
#[derive(Debug, Default)]
pub struct BufferChainBuilder<T> {
    elems: SmallVec<[BufferElement; 16]>,
    split: usize,
    marker: PhantomData<T>,
}

impl BufferChainBuilder<Readable> {
    /// Create a new builder in the [`Readable`] state.
    pub fn new() -> Self {
        Self {
            elems: Default::default(),
            split: 0,
            marker: PhantomData,
        }
    }

    /// Add a readable buffer (device reads from this).
    pub fn readable(mut self, addr: u64, len: u32) -> Self {
        self.elems.push(BufferElement {
            addr,
            len,
            writable: false,
        });
        self.split += 1;
        self
    }

    /// Add multiple readable buffers from an iterator.
    pub fn readables(
        mut self,
        elements: impl IntoIterator<Item = impl Into<BufferElement>>,
    ) -> Self {
        for elem in elements {
            let mut elem = elem.into();
            elem.writable = false;
            self.elems.push(elem);
            self.split += 1;
        }

        self
    }

    /// Add a writable buffer (device writes to this).
    ///
    /// This transitions to Writable state so no more readable buffers can be added.
    pub fn writable(mut self, addr: u64, len: u32) -> BufferChainBuilder<Writable> {
        self.elems.push(BufferElement {
            addr,
            len,
            writable: true,
        });

        BufferChainBuilder {
            elems: self.elems,
            split: self.split,
            marker: PhantomData,
        }
    }

    /// Add multiple writable buffers from an iterator.
    ///
    /// This transitions to Writable state so no more readable buffers can be added.
    pub fn writables(
        mut self,
        elements: impl IntoIterator<Item = impl Into<BufferElement>>,
    ) -> BufferChainBuilder<Writable> {
        for elem in elements {
            let mut elem = elem.into();
            elem.writable = true;
            self.elems.push(elem);
        }

        BufferChainBuilder {
            elems: self.elems,
            split: self.split,
            marker: PhantomData,
        }
    }

    /// Build a buffer chain with only readable buffers.
    ///
    /// Chain must have at least one buffer otherwise an error is returned.
    pub fn build(self) -> Result<BufferChain, RingError> {
        if self.elems.is_empty() {
            return Err(RingError::EmptyChain);
        }

        Ok(BufferChain {
            elems: self.elems,
            split: self.split,
        })
    }
}

impl BufferChainBuilder<Writable> {
    /// Add writable buffer
    pub fn writable(mut self, addr: u64, len: u32) -> Self {
        self.elems.push(BufferElement {
            addr,
            len,
            writable: true,
        });
        self
    }

    /// Add multiple writable buffers from an iterator.
    pub fn writables(
        mut self,
        elements: impl IntoIterator<Item = impl Into<BufferElement>>,
    ) -> Self {
        for elem in elements {
            let mut elem = elem.into();
            elem.writable = true;
            self.elems.push(elem);
        }
        self
    }

    /// Build the buffer chain.
    ///
    /// Chain must have at least one buffer otherwise an error is returned.
    pub fn build(self) -> Result<BufferChain, RingError> {
        if self.elems.is_empty() {
            return Err(RingError::EmptyChain);
        }

        Ok(BufferChain {
            elems: self.elems,
            split: self.split,
        })
    }
}

/// A chain of buffers ready for submission to the virtqueue.
///
/// Contains a scatter-gather list of [`BufferElement`]s, divided into
/// readable (driver->device) and writable (device->driver) sections.
#[derive(Debug, Clone)]
pub struct BufferChain {
    /// All buffer elements (readable followed by writable)
    elems: SmallVec<[BufferElement; 16]>,
    /// Split index between readable and writable buffers
    split: usize,
}

impl BufferChain {
    /// Get all buffer elements in the chain.
    pub fn elems(&self) -> &[BufferElement] {
        self.elems.as_slice()
    }

    /// Get readable buffers in chain
    pub fn readables(&self) -> &[BufferElement] {
        &self.elems[..self.split]
    }

    /// Get mutable readable buffers in chain.
    pub(crate) fn readables_mut(&mut self) -> &mut [BufferElement] {
        &mut self.elems[..self.split]
    }

    /// Get writable buffers in chain
    pub fn writables(&self) -> &[BufferElement] {
        &self.elems[self.split..]
    }

    /// Get total number of buffers in chain
    // Note: buffer chain cannot be empty by construction
    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> usize {
        self.elems.len()
    }
}

/// Tracks position in a ring buffer with wrap-around handling.
///
/// The cursor maintains both an index into the ring and a wrap counter
/// that toggles each time the index wraps around.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub struct RingCursor {
    head: u16,
    size: u16,
    wrap: bool,
}

impl RingCursor {
    pub(crate) fn new(size: usize) -> Self {
        Self {
            head: 0,
            size: size as u16,
            wrap: true,
        }
    }

    /// Advance to next position, wrapping around and toggling wrap counter if needed
    #[inline]
    fn advance(&mut self) {
        debug_assert!(self.head.checked_add(1).is_some());
        self.head += 1;
        if self.head >= self.size {
            self.head = 0;
            self.wrap = !self.wrap;
        }
    }

    /// Advance by n positions using modular arithmetic.
    #[inline]
    fn advance_by(&mut self, n: u16) {
        debug_assert!(self.head.checked_add(n).is_some());
        let new = self.head + n;
        let wraps = new / self.size;
        self.head = new % self.size;
        if !wraps.is_multiple_of(2) {
            self.wrap = !self.wrap;
        }
    }

    /// Get current head index
    #[inline]
    pub fn head(&self) -> u16 {
        self.head
    }

    /// Get current wrap counter
    #[inline]
    pub fn wrap(&self) -> bool {
        self.wrap
    }

    /// Reset cursor to initial state.
    #[inline]
    pub fn reset(&mut self) {
        self.head = 0;
        self.wrap = true;
    }
}

/// Producer (driver) side of a packed virtqueue.
///
/// The producer submits buffer chains for the device to process and polls
/// for completions. This is typically used by the driver/guest side.
///
/// # Lifecycle
///
/// 1.Submit: Call [`submit_available`](Self::submit_available) or
///    [`submit_one`](Self::submit_one) to make buffers available to device
/// 2. Notify: If `SubmitResult::notify` is true, signal the device
/// 3. Poll: Call [`poll_used`](Self::poll_used) to check for completions
/// 4. Process: Handle completed buffers and reuse descriptor IDs
#[derive(Debug)]
pub struct RingProducer<M> {
    /// Memory accessor
    mem: M,
    /// Next available descriptor position
    avail_cursor: RingCursor,
    /// Next used descriptor position
    used_cursor: RingCursor,
    /// Free slots in the ring
    num_free: usize,
    /// Descriptor table in shared memory
    desc_table: DescTable,
    /// Shadow of driver event flags (last written value)
    event_flags_shadow: EventFlags,
    // controls when device notifies about used buffers
    drv_evt_addr: u64,
    // reads device event to check if device wants notification
    dev_evt_addr: u64,
    /// stack of free IDs, allows out-of-order completion
    id_free: SmallVec<[u16; DescTable::DEFAULT_LEN]>,
    // chain length per ID, index = ID,
    id_num: SmallVec<[u16; DescTable::DEFAULT_LEN]>,
}

impl<M: MemOps> RingProducer<M> {
    /// Create a new producer from a memory layout and accessor.
    pub fn new(layout: Layout, mem: M) -> Self {
        let size = layout.desc_table_len() as usize;
        let raw = layout.desc_table_addr();

        // SAFETY: Layout fields are private and from_base validates ring geometry.
        let table = unsafe { DescTable::from_raw_parts(raw, size) };
        let cursor = RingCursor::new(size);

        const DEFAULT_LEN: usize = DescTable::default_len();
        let id_free = (0..size as u16).collect::<SmallVec<[_; DEFAULT_LEN]>>();
        let id_num = SmallVec::<[_; DEFAULT_LEN]>::from_elem(0, size);

        // Notification enabled by default
        let event_flags_shadow = EventFlags::ENABLE;

        Self {
            mem,
            avail_cursor: cursor,
            used_cursor: cursor,
            num_free: size,
            desc_table: table,
            id_free,
            id_num,
            event_flags_shadow,
            drv_evt_addr: layout.drv_evt_addr(),
            dev_evt_addr: layout.dev_evt_addr(),
        }
    }

    /// Fast path: submit exactly one descriptor
    ///
    /// This is more efficient than [`submit_available`](Self::submit_available)
    /// for single-buffer submissions as it avoids chain iteration overhead.
    ///
    /// # Arguments
    ///
    /// * `addr` - physical address of the buffer
    /// * `len` - Length of the buffer in bytes
    /// * `writable` - If true, device writes to buffer; if false, device reads
    ///
    /// # Returns
    ///
    /// The descriptor ID assigned to this buffer, for matching with completions.
    ///
    /// # Errors
    ///
    /// - [`RingError::WouldBlock`] - No free descriptor slots
    /// - [`RingError::OutOfMemory`] - No free descriptor IDs (internal error)
    /// - [`RingError::InvalidState`] - ID tracking corrupted (internal error)
    /// - [`RingError::MemError`] - Backend memory error while publishing the descriptor
    pub fn submit_one(&mut self, addr: u64, len: u32, writable: bool) -> Result<u16, RingError> {
        if self.num_free < 1 {
            return Err(RingError::WouldBlock);
        }

        // Allocate ID and record chain length
        let id = self.id_free.pop().ok_or(RingError::OutOfMemory)?;

        // We should never reuse an ID that is still outstanding
        if self.id_num[id as usize] != 0 {
            return Err(RingError::InvalidState);
        }

        // Record chain length for single descriptor
        self.id_num[id as usize] = 1;

        // Build and publish the head descriptor
        let head_idx = self.avail_cursor.head();
        let head_wrap = self.avail_cursor.wrap();

        let mut flags = DescFlags::empty();
        flags.set(DescFlags::WRITE, writable);
        let mut desc = Descriptor::new(addr, len, id, flags);
        desc.mark_avail(head_wrap);

        let addr = self
            .desc_table
            .desc_addr(head_idx)
            .ok_or(RingError::InvalidState)?;

        // Release publish
        desc.write_release(&self.mem, addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteDesc, addr))?;

        // Advance state
        self.avail_cursor.advance();
        self.num_free -= 1;

        Ok(id)
    }

    /// Submit a buffer chain to the ring, returning whether to notify the device.
    pub fn submit_available_with_notify(
        &mut self,
        chain: &BufferChain,
    ) -> Result<SubmitResult, RingError> {
        let old = self.avail_cursor;
        let id = self.submit_available(chain)?;
        let new = self.avail_cursor;
        let notify = self.should_notify_device(old, new)?;

        Ok(SubmitResult { id, notify })
    }

    /// Submit a single-buffer descriptor with notification check.
    pub fn submit_one_with_notify(
        &mut self,
        addr: u64,
        len: u32,
        writable: bool,
    ) -> Result<SubmitResult, RingError> {
        let old = self.avail_cursor;
        let id = self.submit_one(addr, len, writable)?;
        let new = self.avail_cursor;
        let notify = self.should_notify_device(old, new)?;
        Ok(SubmitResult { id, notify })
    }

    /// Submit a buffer chain to the ring.
    ///
    /// Writes all descriptors in the chain to the ring, linking them with
    /// NEXT flags. The head descriptor is written last with release semantics
    /// to ensure atomicity of the chain.
    ///
    /// # Arguments
    ///
    /// * `chain` - The buffer chain to submit
    ///
    /// # Returns
    ///
    /// The descriptor ID assigned to this chain. All descriptors in the chain
    /// share this ID for correlation during completion.
    ///
    /// # Errors
    ///
    /// - [`RingError::EmptyChain`] - Chain has no buffers
    /// - [`RingError::WouldBlock`] - Not enough free descriptor slots
    /// - [`RingError::OutOfMemory`] - No free descriptor IDs (internal error)
    /// - [`RingError::InvalidState`] - ID tracking or descriptor-table state is corrupted
    /// - [`RingError::MemError`] - Backend memory error while publishing descriptors
    pub fn submit_available(&mut self, chain: &BufferChain) -> Result<u16, RingError> {
        let total_descs = chain.len();
        if total_descs == 0 {
            return Err(RingError::EmptyChain);
        }

        if self.num_free < total_descs {
            return Err(RingError::WouldBlock);
        }

        if total_descs == 1 {
            let elem = chain.elems()[0];
            return self.submit_one(elem.addr, elem.len, elem.writable);
        }

        let head_idx = self.avail_cursor.head();
        let head_wrap = self.avail_cursor.wrap();

        let id = self.id_free.pop().ok_or(RingError::OutOfMemory)?;

        // We should never reuse an ID that is still outstanding
        if self.id_num[id as usize] != 0 {
            return Err(RingError::InvalidState);
        }

        // Record chain length
        self.id_num[id as usize] = total_descs as u16;

        // Write tail elements first; head last.
        let mut pos = self.avail_cursor;
        pos.advance();

        for (i, elem) in chain.elems().iter().enumerate().skip(1) {
            let is_next = i + 1 < total_descs;
            let mut flags = DescFlags::empty();

            flags.set(DescFlags::NEXT, is_next);
            flags.set(DescFlags::WRITE, elem.writable);

            let mut desc = Descriptor::new(elem.addr, elem.len, id, flags);
            desc.mark_avail(pos.wrap());

            let addr = self
                .desc_table
                .desc_addr(pos.head())
                .ok_or(RingError::InvalidState)?;

            self.mem
                .write_val(addr, desc)
                .map_err(|_| RingError::mem_err(MemOp::WriteDesc, addr))?;
            pos.advance();
        }

        // Head descriptor
        let head_elem = chain.elems()[0];
        // Record chain length
        let mut head_flags = DescFlags::empty();
        head_flags.set(DescFlags::NEXT, total_descs > 1);
        head_flags.set(DescFlags::WRITE, head_elem.writable);

        let mut head_desc = Descriptor::new(head_elem.addr, head_elem.len, id, head_flags);
        head_desc.mark_avail(head_wrap);

        let head_addr = self
            .desc_table
            .desc_addr(head_idx)
            .ok_or(RingError::InvalidState)?;

        // Release publish
        head_desc
            .write_release(&self.mem, head_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteDesc, head_addr))?;

        self.num_free -= total_descs;
        self.avail_cursor = pos;

        Ok(id)
    }

    /// Poll the ring for a used buffer.
    ///
    /// Checks if the device has marked any buffers as used. If so, returns
    /// the completion information and reclaims the descriptor(s).
    ///
    /// # Returns
    ///
    /// - `Ok(UsedBuffer)` - A buffer chain was completed
    /// - `Err(RingError::WouldBlock)` - No completions available
    pub fn poll_used(&mut self) -> Result<UsedBuffer, RingError> {
        let idx = self.used_cursor.head();
        let wrap = self.used_cursor.wrap();

        // Read the descriptor at next_used position with ordering
        let addr = self
            .desc_table
            .desc_addr(idx)
            .ok_or(RingError::InvalidState)?;

        // Acquire flags then fields (publish point)
        let flags = Descriptor::read_flags_acquire(&self.mem, addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadDesc, addr))?;

        if !flags.is_used(wrap) {
            return Err(RingError::WouldBlock);
        }

        let desc = Descriptor::read_body(&self.mem, addr, flags)
            .map_err(|_| RingError::mem_err(MemOp::ReadDesc, addr))?;

        let id = desc.id;
        let count = *self
            .id_num
            .get(id as usize)
            .ok_or(RingError::InvalidState)?;

        if count == 0 {
            return Err(RingError::InvalidState);
        }

        // Advance used cursor by number of reclaimed descriptors
        self.used_cursor.advance_by(count);
        // Update number of free descriptors
        self.num_free += count as usize;
        // SAFETY: id is valid because we checked above
        self.id_num[id as usize] = 0;
        // Return ID to free stack
        self.id_free.push(id);

        Ok(UsedBuffer { id, len: desc.len })
    }

    /// Get number of free descriptors in the ring.
    #[inline]
    pub fn num_free(&self) -> usize {
        self.num_free
    }

    /// Get number of inflight (submitted but not yet used) descriptors.
    #[inline]
    pub fn num_inflight(&self) -> usize {
        self.desc_table.len() - self.num_free
    }

    /// Check if the ring is full (no free descriptors).
    #[inline]
    pub fn is_full(&self) -> bool {
        self.num_free == 0
    }

    /// Get descriptor table length
    #[inline]
    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> usize {
        self.desc_table.len()
    }

    /// Get memory accessor reference
    #[inline]
    pub fn mem(&self) -> &M {
        &self.mem
    }

    /// Get descriptor table reference
    #[inline]
    pub fn desc_table(&self) -> &DescTable {
        &self.desc_table
    }

    /// Get a snapshot of the current available cursor position.
    ///
    /// Used for batch operations to track the cursor before submitting
    /// multiple chains, enabling proper event suppression checks.
    #[inline]
    pub fn avail_cursor(&self) -> RingCursor {
        self.avail_cursor
    }

    /// Get a snapshot of the current used cursor position.
    ///
    /// Used for setting up DESC mode event suppression at specific positions.
    #[inline]
    pub fn used_cursor(&self) -> RingCursor {
        self.used_cursor
    }

    /// Check if device should be notified given a cursor snapshot from before batch start.
    ///
    /// This is used for batching: record cursor before first submit, then after all
    /// submits call this to determine if notification is needed based on event suppression.
    ///
    /// # Arguments
    /// * `old` - Cursor position snapshot taken before batch started
    pub fn should_notify_since(&self, old: RingCursor) -> Result<bool, RingError> {
        self.should_notify_device(old, self.avail_cursor)
    }

    /// Driver disables used-buffer notifications from device to driver.
    pub fn disable_used_notifications(&mut self) -> Result<(), RingError> {
        // Avoid redundant MMIO writes if already disabled
        if self.event_flags_shadow == EventFlags::DISABLE {
            return Ok(());
        }

        let mut evt = self
            .mem
            .read_val::<EventSuppression>(self.drv_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, self.drv_evt_addr))?;

        evt.set_flags(EventFlags::DISABLE);

        evt.write_release(&self.mem, self.drv_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteEvent, self.drv_evt_addr))?;
        self.event_flags_shadow = EventFlags::DISABLE;
        Ok(())
    }

    /// Driver enables used-buffer notifications from device to driver.
    pub fn enable_used_notifications(&mut self) -> Result<(), RingError> {
        if self.event_flags_shadow == EventFlags::ENABLE {
            return Ok(());
        }

        let mut evt = self
            .mem
            .read_val::<EventSuppression>(self.drv_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, self.drv_evt_addr))?;

        evt.set_flags(EventFlags::ENABLE);
        evt.write_release(&self.mem, self.drv_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteEvent, self.drv_evt_addr))?;

        self.event_flags_shadow = EventFlags::ENABLE;
        Ok(())
    }

    /// Driver enables descriptor-specific used notifications (EVENT_IDX / DESC mode).
    ///
    /// This tells the device: "Interrupt me when you reach used index (off, wrap)".
    ///
    /// This enables batching on the device side - it can complete multiple requests
    /// before triggering an interrupt.
    pub fn enable_used_notifications_desc(
        &mut self,
        off: u16,
        wrap: bool,
    ) -> Result<(), RingError> {
        let mut evt = self
            .mem
            .read_val::<EventSuppression>(self.drv_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, self.drv_evt_addr))?;

        evt.set_desc_event(off, wrap);
        evt.set_flags(EventFlags::DESC);

        // Now publish flags = DESC with Release semantics.
        evt.write_release(&self.mem, self.drv_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteEvent, self.drv_evt_addr))?;
        // cache shadow
        self.event_flags_shadow = EventFlags::DESC;
        Ok(())
    }

    /// Convenience: enable DESC mode for "next used cursor" like Linux enable_cb_prepare.
    pub fn enable_used_notifications_for_next(&mut self) -> Result<(), RingError> {
        let off = self.used_cursor.head();
        let wrap = self.used_cursor.wrap();

        self.enable_used_notifications_desc(off, wrap)
    }

    /// Check whether the device should be notified about new available descriptors.
    fn should_notify_device(&self, old: RingCursor, new: RingCursor) -> Result<bool, RingError> {
        // VIRTIO 1.1 "The driver MUST perform a suitable memory barrier before
        // reading the Device Event Suppression structure".
        //
        // After publishing descriptors with store-release on the AVAIL/USED flags,
        // we need a full barrier before reading event suppression, because
        // release+acquire across different memory locations does NOT provide
        // Store/Load ordering on weakly-ordered architectures e.g. aarch64.
        //
        // Linux kernel uses virtio_mb() full barrier in virtqueue_kick_prepare_packed.
        fence(Ordering::SeqCst);

        should_notify_evt(&self.mem, self.dev_evt_addr, self.len() as u16, old, new)
    }

    /// Reset to initial state matching a freshly zeroed ring.
    pub fn reset(&mut self) {
        let size = self.desc_table.len();
        self.avail_cursor.reset();
        self.used_cursor.reset();
        self.num_free = size;
        self.id_free.clear();
        self.id_free.extend(0..size as u16);
        self.id_num.iter_mut().for_each(|n| *n = 0);
        self.event_flags_shadow = EventFlags::ENABLE;
    }

    /// Reset the ring to the "N slots submitted, none completed" state.
    ///
    /// `ids` contains the descriptor IDs that are in-flight.
    /// Sets cursors, counters, and `id_num` accordingly. The chain lengths are all set to 1.
    pub fn reset_prefilled(&mut self, ids: &[u16]) {
        let size = self.desc_table.len();
        let count = ids.len();
        assert!(count <= size);

        let wrapped = count >= size;
        self.avail_cursor.head = if wrapped { 0 } else { count as u16 };
        self.avail_cursor.wrap = !wrapped;

        self.used_cursor.head = 0;
        self.used_cursor.wrap = true;

        self.id_num.iter_mut().for_each(|n| *n = 0);
        for &id in ids {
            assert!((id as usize) < size);
            assert_eq!(self.id_num[id as usize], 0);
            self.id_num[id as usize] = 1;
        }

        self.num_free = size - count;
        self.id_free.clear();
        self.id_free
            .extend((0..size as u16).filter(|id| self.id_num[*id as usize] == 0));
    }
}

/// Consumer (device) side of a packed virtqueue.
///
/// The consumer polls for available buffer chains submitted by the driver,
/// processes them, and marks them as used. This is typically used by the
/// device/host side.
///
/// # Lifecycle
///
/// 1. **Poll**: Call [`poll_available`](Self::poll_available) to get buffers
/// 2. **Process**: Read from readable buffers, write to writable buffers
/// 3. **Complete**: Call [`submit_used`](Self::submit_used) to return buffers
/// 4. **Notify**: If `submit_used_with_notify` returns true, signal the driver
#[derive(Debug)]
pub struct RingConsumer<M> {
    /// Memory accessor
    mem: M,
    /// Cursor for reading available (driver-published) descriptors
    avail_cursor: RingCursor,
    /// Cursor for writing used descriptors
    used_cursor: RingCursor,
    /// Shared descriptor table
    desc_table: DescTable,
    /// Per-ID chain length learned when polling (index = ID)
    id_num: SmallVec<[u16; DescTable::DEFAULT_LEN]>,
    /// Number of descriptors consumed from avail stream but not yet posted as used.
    num_inflight: usize,
    /// Shadow of device event flags (last written value)
    event_flags_shadow: EventFlags,
    // reads driver event to control when device should notify
    drv_evt_addr: u64,
    // write device_event (checks if device wants notification about available buffers)
    dev_evt_addr: u64,
}

impl<M: MemOps> RingConsumer<M> {
    pub fn new(layout: Layout, mem: M) -> Self {
        let size = layout.desc_table_len() as usize;
        let raw = layout.desc_table_addr();

        // SAFETY: Layout fields are private and from_base validates ring geometry.
        let table = unsafe { DescTable::from_raw_parts(raw, size) };
        let cursor = RingCursor::new(size);
        let id_chain_len = SmallVec::<[u16; DescTable::DEFAULT_LEN]>::from_elem(0, size);

        // Notification enabled by default
        let event_flags_shadow = EventFlags::ENABLE;

        Self {
            mem,
            avail_cursor: cursor,
            used_cursor: cursor,
            desc_table: table,
            id_num: id_chain_len,
            num_inflight: 0,
            event_flags_shadow,
            drv_evt_addr: layout.drv_evt_addr(),
            dev_evt_addr: layout.dev_evt_addr(),
        }
    }

    /// Poll for an available buffer chain.
    ///
    /// Returns the chain ID and a [`BufferChain`] containing all buffers.
    /// The chain ID must be passed to [`submit_used`](Self::submit_used)
    /// when processing is complete.
    ///
    /// # Returns
    ///
    /// - `Ok((id, chain))` - A buffer chain is available
    /// - `Err(RingError::WouldBlock)` - No buffers available
    /// - `Err(RingError::BadChain)` - Malformed chain (driver bug)
    pub fn poll_available(&mut self) -> Result<(u16, BufferChain), RingError> {
        let idx = self.avail_cursor.head();
        let wrap = self.avail_cursor.wrap();

        let head_addr = self
            .desc_table
            .desc_addr(idx)
            .ok_or(RingError::InvalidState)?;

        // Acquire: flags then fields (publish point)
        let flags = Descriptor::read_flags_acquire(&self.mem, head_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadDesc, head_addr))?;

        // Check if head descriptor is available to consume
        if !flags.is_avail(wrap) {
            return Err(RingError::WouldBlock);
        }

        let head_desc = Descriptor::read_body(&self.mem, head_addr, flags)
            .map_err(|_| RingError::mem_err(MemOp::ReadDesc, head_addr))?;

        if flags.contains(DescFlags::INDIRECT) {
            return Err(RingError::BadChain);
        }

        // Build chain (head + tails), tracking readable/writable split inline.
        let mut elements = SmallVec::<[BufferElement; 16]>::new();
        let mut pos = self.avail_cursor;
        let mut chain_len: u16 = 1;

        let mut steps = 1;
        let mut has_next = head_desc.is_next();

        let max_steps = self.desc_table.len();

        let head_elem = BufferElement::from(&head_desc);
        let mut seen_writable = head_elem.writable;
        let mut writables: usize = if seen_writable { 1 } else { 0 };
        elements.push(head_elem);
        pos.advance();

        while has_next && steps < max_steps {
            let addr = self
                .desc_table
                .desc_addr(pos.head())
                .ok_or(RingError::InvalidState)?;

            // tail reads does not need ordering because head has been already validated
            let desc: Descriptor = self
                .mem
                .read_val(addr)
                .map_err(|_| RingError::mem_err(MemOp::ReadDesc, addr))?;
            if desc.flags().contains(DescFlags::INDIRECT) {
                return Err(RingError::BadChain);
            }

            let elem = BufferElement::from(&desc);

            if elem.writable {
                seen_writable = true;
                writables += 1;
            } else if seen_writable {
                return Err(RingError::BadChain);
            }

            elements.push(elem);

            chain_len += 1;
            steps += 1;

            has_next = desc.is_next();
            pos.advance();
        }

        // Detect malformed chains, this means we reached max_steps but still have NEXT set.
        if steps >= max_steps && has_next {
            return Err(RingError::BadChain);
        }

        // Check if next inflight will exceed ring capacity - this should never happen if driver is
        // well-behaved and we correctly track inflight count.
        if self.num_inflight + chain_len as usize > self.desc_table.len() {
            return Err(RingError::InvalidState);
        }

        let readables = elements.len() - writables;

        // Since driver wrote the same id everywhere, head_desc.id is valid.
        let id = head_desc.id;
        let id_num = self
            .id_num
            .get_mut(id as usize)
            .ok_or(RingError::InvalidState)?;
        if *id_num != 0 {
            return Err(RingError::InvalidState);
        }

        // Record chain length for later used submission
        *id_num = chain_len;
        // Advance avail cursor to first slot after chain
        self.avail_cursor = pos;
        // Update inflight count
        self.num_inflight += chain_len as usize;

        Ok((
            id,
            BufferChain {
                elems: elements,
                split: readables,
            },
        ))
    }

    /// Publish a single used descriptor for the chain identified by id.
    /// written_len is the total bytes produced by the device (for writable part).
    ///
    /// # Arguments
    ///
    /// * `id` - The chain ID from `poll_available`
    /// * `written_len` - Total bytes written to writable buffers
    ///
    /// # Errors
    ///
    /// - [`RingError::InvalidState`] - Unknown ID or already completed
    pub fn submit_used(&mut self, id: u16, written_len: u32) -> Result<(), RingError> {
        // Lookup chain length
        let chain_len = *self
            .id_num
            .get(id as usize)
            .ok_or(RingError::InvalidState)?;

        if chain_len == 0 || chain_len > self.desc_table.len() as u16 {
            return Err(RingError::InvalidState);
        }

        let idx = self.used_cursor.head();
        let wrap = self.used_cursor.wrap();

        // addr is unused for used descriptor according to packed-virtqueue spec
        let mut used_desc = Descriptor::new(0, written_len, id, DescFlags::empty());
        used_desc.mark_used(wrap);

        let addr = self
            .desc_table
            .desc_addr(idx)
            .ok_or(RingError::InvalidState)?;

        // Release publish (flags written last inside write_release)
        used_desc
            .write_release(&self.mem, addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteDesc, addr))?;

        // Advance used cursor by whole chain length
        self.used_cursor.advance_by(chain_len);
        self.id_num[id as usize] = 0;

        self.num_inflight -= chain_len as usize;
        Ok(())
    }

    /// Try to peek whether the next chain is available without consuming it.
    pub fn peek_available(&self) -> Result<bool, RingError> {
        let Some(addr) = self.desc_table.desc_addr(self.avail_cursor.head()) else {
            return Err(RingError::InvalidState);
        };

        let flags = Descriptor::read_flags_acquire(&self.mem, addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadDesc, addr))?;

        Ok(flags.is_avail(self.avail_cursor.wrap()))
    }

    /// Submit a used descriptor and return whether to notify the driver.
    pub fn submit_used_with_notify(
        &mut self,
        id: u16,
        written_len: u32,
    ) -> Result<bool, RingError> {
        let old = self.used_cursor;
        self.submit_used(id, written_len)?;
        let new = self.used_cursor;
        self.should_notify_driver(old, new)
    }

    /// Get number of free descriptors in the ring.
    pub fn num_free(&self) -> usize {
        self.desc_table.len() - self.num_inflight
    }

    /// Get number of inflight (submitted but not yet used) descriptors.
    pub fn num_inflight(&self) -> usize {
        self.num_inflight
    }

    /// Check if the ring is full (no free descriptors).
    pub fn is_full(&self) -> bool {
        self.num_inflight == self.desc_table.len()
    }

    /// Get descriptor table length
    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> usize {
        self.desc_table.len()
    }

    /// Get memory accessor reference
    pub fn mem(&self) -> &M {
        &self.mem
    }

    /// Get a snapshot of the current avail cursor position.
    #[inline]
    pub fn avail_cursor(&self) -> RingCursor {
        self.avail_cursor
    }

    /// Get a snapshot of the current used cursor position.
    #[inline]
    pub fn used_cursor(&self) -> RingCursor {
        self.used_cursor
    }

    /// Device disables available-buffer notifications from driver to device.
    ///
    /// This is the device-side mirror of "disable callbacks" but for avail kicks.
    pub fn disable_avail_notifications(&mut self) -> Result<(), RingError> {
        if self.event_flags_shadow == EventFlags::DISABLE {
            return Ok(());
        }

        let mut evt = self
            .mem
            .read_val::<EventSuppression>(self.dev_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, self.dev_evt_addr))?;

        evt.set_flags(EventFlags::DISABLE);
        evt.write_release(&self.mem, self.dev_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteEvent, self.dev_evt_addr))?;

        self.event_flags_shadow = EventFlags::DISABLE;
        Ok(())
    }

    /// Device enables available-buffer notifications from driver to device.
    pub fn enable_avail_notifications(&mut self) -> Result<(), RingError> {
        if self.event_flags_shadow == EventFlags::ENABLE {
            return Ok(());
        }

        let mut evt = self
            .mem
            .read_val::<EventSuppression>(self.dev_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, self.dev_evt_addr))?;

        evt.set_flags(EventFlags::ENABLE);
        evt.write_release(&self.mem, self.dev_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteEvent, self.dev_evt_addr))?;

        self.event_flags_shadow = EventFlags::ENABLE;
        Ok(())
    }

    /// Device enables descriptor-specific available notifications (EVENT_IDX / DESC mode).
    ///
    /// This tells the driver: "Kick me when you reach avail index (off, wrap)".
    pub fn enable_avail_notifications_desc(
        &mut self,
        off: u16,
        wrap: bool,
    ) -> Result<(), RingError> {
        // Update off_wrap first
        let mut evt = self
            .mem
            .read_val::<EventSuppression>(self.dev_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, self.dev_evt_addr))?;

        evt.set_desc_event(off, wrap);
        evt.set_flags(EventFlags::DESC);

        // Now publish flags = DESC with Release semantics.
        evt.write_release(&self.mem, self.dev_evt_addr)
            .map_err(|_| RingError::mem_err(MemOp::WriteEvent, self.dev_evt_addr))?;

        self.event_flags_shadow = EventFlags::DESC;
        Ok(())
    }

    /// Convenience: enable DESC mode for "next avail cursor" (device wants a kick when new
    /// buffers arrive at the next index it will poll).
    pub fn enable_avail_notifications_for_next(&mut self) -> Result<(), RingError> {
        let off = self.avail_cursor.head();
        let wrap = self.avail_cursor.wrap();
        self.enable_avail_notifications_desc(off, wrap)
    }

    /// Decide whether the device should notify the driver about newly used descriptors.
    fn should_notify_driver(&self, old: RingCursor, new: RingCursor) -> Result<bool, RingError> {
        // VIRTIO 1.1: Full memory barrier required before reading the
        // Driver Event Suppression structure. See also should_notify_device()
        fence(Ordering::SeqCst);

        should_notify_evt(&self.mem, self.drv_evt_addr, self.len() as u16, old, new)
    }

    /// Reset to initial state matching a freshly zeroed ring.
    /// Does not reallocate internal buffers.
    pub fn reset(&mut self) {
        self.avail_cursor.reset();
        self.used_cursor.reset();
        self.id_num.iter_mut().for_each(|n| *n = 0);
        self.num_inflight = 0;
        self.event_flags_shadow = EventFlags::ENABLE;
    }
}

/// Read an event-suppression structure and decide whether to notify the peer.
fn should_notify_evt<M: MemOps>(
    mem: &M,
    evt_addr: u64,
    ring_len: u16,
    old: RingCursor,
    new: RingCursor,
) -> Result<bool, RingError> {
    let flags = EventSuppression::read_flags_acquire(mem, evt_addr)
        .map_err(|_| RingError::mem_err(MemOp::ReadEvent, evt_addr))?;

    let evt = if flags == EventFlags::DESC {
        EventSuppression::read_body(mem, evt_addr, flags)
            .map_err(|_| RingError::mem_err(MemOp::ReadEvent, evt_addr))?
    } else {
        EventSuppression::new(0, flags)
    };

    Ok(should_notify(evt, ring_len, old, new))
}

/// Common packed-ring notification decision:
/// - `old` and `new` are the ring indices (head) before/after publishing a batch
/// - `new.wrap()` is the wrap counter corresponding to `new.head()`
/// - `evt.desc_event_wrap()` is compared against `new.wrap()`
///
/// This is compatible with Linux `virtqueue_kick_prepare_packed` logic
#[inline]
fn should_notify(evt: EventSuppression, ring_len: u16, old: RingCursor, new: RingCursor) -> bool {
    match evt.flags() {
        EventFlags::DISABLE => false,
        EventFlags::ENABLE => true,
        EventFlags::DESC => {
            let mut off = evt.desc_event_off();
            if off >= ring_len {
                return false;
            }

            let wrap = evt.desc_event_wrap();

            if wrap != new.wrap() {
                off = off.wrapping_sub(ring_len);
            }

            ring_need_event(off, new.head(), old.head())
        }
        // treat as disabled if invalid
        _ => false,
    }
}

#[inline(always)]
fn ring_need_event(event_idx: u16, new: u16, old: u16) -> bool {
    new.wrapping_sub(event_idx).wrapping_sub(1) < new.wrapping_sub(old)
}

impl From<&Descriptor> for BufferElement {
    fn from(desc: &Descriptor) -> Self {
        BufferElement {
            addr: desc.addr,
            len: desc.len,
            writable: desc.is_writable(),
        }
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use alloc::sync::Arc;
    use core::cell::UnsafeCell;
    use core::num::NonZeroU16;
    use core::ptr;
    use core::sync::atomic::{AtomicU16, Ordering};

    use bytemuck::{Pod, Zeroable};

    use super::super::align_up;
    use super::*;
    use crate::virtq::event::EventSuppression;

    /// Test MemOps implementation that maintains pointer provenance.
    ///
    /// Wraps shared storage behind Arc for cheap cloning. This allows
    /// producer and consumer to share the same backing memory without
    /// Arc appearing in the type signatures.
    #[derive(Clone)]
    pub struct TestMem {
        inner: Arc<TestMemInner>,
    }

    struct TestMemInner {
        /// The backing storage - UnsafeCell for interior mutability
        storage: UnsafeCell<Vec<u8>>,
        /// Base address (the address we tell the ring about)
        base_addr: u64,
    }

    // Safety: TestMemInner's UnsafeCell is only accessed from test code
    // with no real concurrency in unit tests (loom tests use LoomMem).
    unsafe impl Send for TestMemInner {}
    unsafe impl Sync for TestMemInner {}

    impl TestMem {
        pub fn new(size: usize) -> Self {
            let storage = vec![0u8; size];
            let base_addr = storage.as_ptr() as u64;
            Self {
                inner: Arc::new(TestMemInner {
                    storage: UnsafeCell::new(storage),
                    base_addr,
                }),
            }
        }

        /// Get a pointer with proper provenance for the given address
        fn ptr_for_addr(&self, addr: u64) -> *mut u8 {
            let storage = unsafe { &mut *self.inner.storage.get() };
            let base_ptr = storage.as_mut_ptr();
            let offset = (addr - self.inner.base_addr) as usize;
            // Use wrapping_add to maintain provenance from base_ptr
            base_ptr.wrapping_add(offset)
        }

        pub fn base_addr(&self) -> u64 {
            self.inner.base_addr
        }
    }

    // SAFETY: TestMem translates addresses into its owned backing storage. Unit
    // tests construct layouts within that storage and avoid concurrent access.
    unsafe impl MemOps for TestMem {
        type Error = core::convert::Infallible;

        fn read(&self, addr: u64, dst: &mut [u8]) -> Result<(), Self::Error> {
            let src = self.ptr_for_addr(addr);
            unsafe {
                ptr::copy_nonoverlapping(src, dst.as_mut_ptr(), dst.len());
            }
            Ok(())
        }

        fn write(&self, addr: u64, src: &[u8]) -> Result<(), Self::Error> {
            let dst = self.ptr_for_addr(addr);
            unsafe {
                ptr::copy_nonoverlapping(src.as_ptr(), dst, src.len());
            }
            Ok(())
        }

        fn read_val<T: Pod>(&self, addr: u64) -> Result<T, Self::Error> {
            let ptr = self.ptr_for_addr(addr).cast::<T>();
            Ok(unsafe { ptr::read_volatile(ptr) })
        }

        fn write_val<T: Pod>(&self, addr: u64, val: T) -> Result<(), Self::Error> {
            let ptr = self.ptr_for_addr(addr).cast::<T>();
            unsafe { ptr::write_volatile(ptr, val) };
            Ok(())
        }

        fn load_acquire(&self, addr: u64) -> Result<u16, Self::Error> {
            let ptr = self.ptr_for_addr(addr).cast::<AtomicU16>();
            Ok(unsafe { (*ptr).load(Ordering::Acquire) })
        }

        fn store_release(&self, addr: u64, val: u16) -> Result<(), Self::Error> {
            let ptr = self.ptr_for_addr(addr).cast::<AtomicU16>();
            unsafe { (*ptr).store(val, Ordering::Release) };
            Ok(())
        }

        unsafe fn as_slice(&self, addr: u64, len: usize) -> Result<&[u8], Self::Error> {
            let ptr = self.ptr_for_addr(addr);
            Ok(unsafe { core::slice::from_raw_parts(ptr, len) })
        }

        unsafe fn as_mut_slice(&self, addr: u64, len: usize) -> Result<&mut [u8], Self::Error> {
            let ptr = self.ptr_for_addr(addr);
            Ok(unsafe { core::slice::from_raw_parts_mut(ptr, len) })
        }
    }

    /// Owns the descriptor table and event suppression structures
    pub struct OwnedRing {
        mem: TestMem,
        layout: Layout,
    }

    impl OwnedRing {
        pub fn new(size: usize) -> Self {
            let num_descs = NonZeroU16::new(size as u16).unwrap();
            let needed = Layout::query_size(size);

            // Add padding for alignment, plus extra space for pool buffers
            // used by high-level API tests (pool offset = ring_end + 0x100,
            // pool size = 0x8000).
            let padding = Descriptor::ALIGN;
            let pool_headroom = 0x100 + 0x8000;
            let mem = TestMem::new(needed + padding + pool_headroom);

            // Align the base address
            let aligned_base = align_up(mem.base_addr() as usize, Descriptor::ALIGN) as u64;
            let layout = unsafe { Layout::from_base(aligned_base, num_descs).unwrap() };

            Self { mem, layout }
        }

        pub fn layout(&self) -> Layout {
            self.layout
        }

        pub fn mem(&self) -> TestMem {
            self.mem.clone()
        }

        /// Get address of descriptor at index
        pub fn desc_addr(&self, idx: u16) -> u64 {
            self.layout.desc_table_addr() + (idx as u64 * Descriptor::SIZE as u64)
        }

        /// Read descriptor directly (for test verification)
        pub fn read_desc(&self, idx: u16) -> Descriptor {
            self.mem.read_val(self.desc_addr(idx)).unwrap()
        }

        /// Write descriptor directly (for test manipulation)
        pub fn write_desc(&self, idx: u16, desc: Descriptor) {
            self.mem.write_val(self.desc_addr(idx), desc).unwrap()
        }

        /// Read driver event directly
        pub fn read_driver_event(&self) -> EventSuppression {
            self.mem.read_val(self.layout.drv_evt_addr()).unwrap()
        }

        /// Read device event directly
        pub fn read_device_event(&self) -> EventSuppression {
            self.mem.read_val(self.layout.dev_evt_addr()).unwrap()
        }

        pub fn len(&self) -> usize {
            self.layout.desc_table_len() as usize
        }
    }

    // Share the TestMem between producer and consumer via reference
    pub(crate) fn make_ring(size: usize) -> OwnedRing {
        OwnedRing::new(size)
    }

    pub(crate) fn make_producer(ring: &OwnedRing) -> RingProducer<TestMem> {
        RingProducer::new(ring.layout(), ring.mem())
    }

    pub(crate) fn make_consumer(ring: &OwnedRing) -> RingConsumer<TestMem> {
        RingConsumer::new(ring.layout(), ring.mem())
    }

    fn assert_invariants(ring: &OwnedRing, prod: &RingProducer<TestMem>) {
        let outstanding: u16 = prod.id_num.iter().copied().sum();
        assert_eq!(outstanding as usize + prod.num_free, ring.len());

        for id in prod.id_free.iter() {
            assert_eq!(prod.id_num[*id as usize], 0);
        }

        for (id, &n) in prod.id_num.iter().enumerate() {
            if n > 0 {
                assert!(!prod.id_free.contains(&(id as u16)));
            }
        }
    }

    #[test]
    fn test_initialization() {
        let ring = make_ring(8);
        let producer = make_producer(&ring);

        // All descriptors should be zeroed
        for i in 0..8u16 {
            let desc = ring.read_desc(i);
            assert_eq!(desc, Descriptor::zeroed());
            assert_eq!(desc.flags, 0);
            assert_eq!(desc.addr, 0);
            assert_eq!(desc.len, 0);
            assert_eq!(desc.id, 0);
        }

        // Cursors start at head=0, wrap=true
        assert_eq!(producer.avail_cursor.head(), 0);
        assert!(producer.avail_cursor.wrap());
        assert_eq!(producer.used_cursor.head(), 0);
        assert!(producer.used_cursor.wrap());

        // All IDs free, id_num zeroed, num_free == size
        assert_eq!(producer.id_free.len(), 8);
        assert_eq!(producer.num_free, 8);
        for i in 0..8 {
            assert_eq!(producer.id_num[i], 0);
        }
    }

    #[test]
    fn test_buffer_chain_builder_normalizes_element_direction() {
        let readable_as_writable = BufferElement {
            addr: 0x1000,
            len: 16,
            writable: true,
        };
        let writable_as_readable = BufferElement {
            addr: 0x2000,
            len: 32,
            writable: false,
        };
        let second_writable_as_readable = BufferElement {
            addr: 0x3000,
            len: 64,
            writable: false,
        };

        let chain = BufferChainBuilder::new()
            .readables([readable_as_writable])
            .writables([writable_as_readable])
            .writables([second_writable_as_readable])
            .build()
            .unwrap();

        assert!(!chain.readables()[0].writable);
        assert!(chain.writables().iter().all(|elem| elem.writable));
    }

    #[test]
    fn test_submit_one_descriptor() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let addr = 0x1000;
        let len = 512;
        let writable = false;

        let id = producer.submit_one(addr, len, writable).unwrap();

        // Check descriptor was written correctly
        let desc = ring.read_desc(0);

        assert_eq!(desc.addr, addr);
        assert_eq!(desc.len, len);
        assert_eq!(desc.id, id);

        // AVAIL should match wrap (true), USED should be inverse (false)
        let flags = desc.flags();
        assert!(flags.contains(DescFlags::AVAIL));
        assert!(!flags.contains(DescFlags::USED));
        assert!(!flags.contains(DescFlags::WRITE));
        assert!(!flags.contains(DescFlags::NEXT));

        // num_free should be decremented
        assert_eq!(producer.num_free, 7);

        // Cursor advanced
        assert_eq!(producer.avail_cursor.head(), 1);
        assert!(producer.avail_cursor.wrap());

        // ID allocated and chain length recorded
        assert_eq!(producer.id_num[id as usize], 1);
        assert_eq!(producer.id_free.len(), 7);
    }

    #[test]
    fn test_single_descriptor_wrap_toggle() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        // Advance to last slot
        producer.avail_cursor.head = 3;
        producer.avail_cursor.wrap = true;
        producer.num_free = 1;
        producer.id_free.clear();
        producer.id_free.push(0);

        let _id = producer.submit_one(0x1000, 512, false).unwrap();

        // After submission, cursor should wrap
        assert_eq!(producer.avail_cursor.head(), 0);
        assert!(!producer.avail_cursor.wrap());

        // Descriptor should have old wrap bits
        let desc = ring.read_desc(3);
        let flags = desc.flags();
        assert!(flags.contains(DescFlags::AVAIL));
        assert!(!flags.contains(DescFlags::USED));
    }

    #[test]
    fn test_multi_descriptor_no_wrap() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .readable(0x2000, 256)
            .writable(0x3000, 512)
            .build()
            .unwrap();

        let id = producer.submit_available(&chain).unwrap();

        // Check head descriptor
        let head_desc = ring.read_desc(0);
        assert_eq!(head_desc.addr, 0x1000);
        assert_eq!(head_desc.len, 256);
        assert_eq!(head_desc.id, id);

        let head_flags = head_desc.flags();
        assert!(head_flags.contains(DescFlags::NEXT));
        assert!(!head_flags.contains(DescFlags::WRITE));
        assert!(head_flags.contains(DescFlags::AVAIL));
        assert!(!head_flags.contains(DescFlags::USED));

        // Check middle descriptor
        let mid_desc = ring.read_desc(1);
        assert_eq!(mid_desc.addr, 0x2000);
        assert_eq!(mid_desc.len, 256);
        assert_eq!(mid_desc.id, id);

        let mid_flags = mid_desc.flags();
        assert!(mid_flags.contains(DescFlags::NEXT));
        assert!(!mid_flags.contains(DescFlags::WRITE));

        // Check tail descriptor
        let tail_desc = ring.read_desc(2);
        assert_eq!(tail_desc.addr, 0x3000);
        assert_eq!(tail_desc.len, 512);
        assert_eq!(tail_desc.id, id);

        let tail_flags = tail_desc.flags();
        assert!(!tail_flags.contains(DescFlags::NEXT));
        assert!(tail_flags.contains(DescFlags::WRITE));

        // All descriptors have same ID
        assert_eq!(head_desc.id, mid_desc.id);
        assert_eq!(mid_desc.id, tail_desc.id);

        // Check state updates
        assert_eq!(producer.num_free, 5);
        assert_eq!(producer.avail_cursor.head(), 3);
        assert_eq!(producer.id_num[id as usize], 3);
    }

    #[test]
    fn test_multi_descriptor_with_wrap() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        // Position head near end
        producer.avail_cursor.head = 2;
        producer.avail_cursor.wrap = true;

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .readable(0x2000, 256)
            .readable(0x3000, 256)
            .build()
            .unwrap();

        let _id = producer.submit_available(&chain).unwrap();

        // Head at index 2 with wrap=true
        let head_desc = ring.read_desc(2);
        let head_flags = head_desc.flags();
        assert!(head_flags.contains(DescFlags::AVAIL));
        assert!(!head_flags.contains(DescFlags::USED));

        // Middle at index 3 with wrap=true (before boundary)
        let mid_desc = ring.read_desc(3);
        let mid_flags = mid_desc.flags();
        assert!(mid_flags.contains(DescFlags::AVAIL));
        assert!(!mid_flags.contains(DescFlags::USED));

        // Tail at index 0 with wrap=false (after boundary)
        let tail_desc = ring.read_desc(0);
        let tail_flags = tail_desc.flags();
        assert!(!tail_flags.contains(DescFlags::AVAIL));
        assert!(tail_flags.contains(DescFlags::USED));

        // Cursor should have wrapped
        assert_eq!(producer.avail_cursor.head(), 1);
        assert!(!producer.avail_cursor.wrap());
    }

    #[test]
    fn test_ring_full() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        // Fill ring completely
        for _ in 0..4 {
            producer.submit_one(0x1000, 256, false).unwrap();
        }

        assert_eq!(producer.num_free, 0);

        // Next submit should fail
        let result = producer.submit_one(0x5000, 256, false);
        assert!(matches!(result, Err(RingError::WouldBlock)));
    }

    #[test]
    fn test_poll_and_reclaim() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let id = producer.submit_one(0x1000, 512, false).unwrap();

        // Manually mark as used (simulate device)
        let mut desc = ring.read_desc(0);
        desc.mark_used(true);
        desc.len = 256;
        ring.write_desc(0, desc);

        // Poll should return the used buffer
        let used = producer.poll_used().unwrap();
        assert_eq!(used.id, id);
        assert_eq!(used.len, 256);

        // State should be updated
        assert_eq!(producer.num_free, 8);
        assert_eq!(producer.used_cursor.head(), 1);
        assert_eq!(producer.id_num[id as usize], 0);
        assert!(producer.id_free.contains(&id));
    }

    #[test]
    fn test_poll_multi_descriptor_chain() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .readable(0x2000, 256)
            .writable(0x3000, 512)
            .build()
            .unwrap();

        let id = producer.submit_available(&chain).unwrap();

        // Mark only head as used
        let mut head_desc = ring.read_desc(0);
        head_desc.mark_used(true);
        head_desc.len = 512;
        ring.write_desc(0, head_desc);

        // Poll should reclaim all 3 descriptors
        let used = producer.poll_used().unwrap();
        assert_eq!(used.id, id);
        assert_eq!(used.len, 512);

        // Should have skipped 3 descriptors
        assert_eq!(producer.used_cursor.head(), 3);
        assert_eq!(producer.num_free, 8);
    }

    #[test]
    fn test_id_reuse() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        // Submit and complete first buffer
        let id1 = producer.submit_one(0x1000, 256, false).unwrap();

        let mut desc = ring.read_desc(0);
        desc.mark_used(true);
        ring.write_desc(0, desc);

        producer.poll_used().unwrap();

        // Submit another buffer - should reuse ID
        let id2 = producer.submit_one(0x2000, 256, false).unwrap();

        // ID should be reused (LIFO from stack)
        assert_eq!(id2, id1);
        assert_eq!(producer.id_num[id2 as usize], 1);
    }

    #[test]
    fn test_available_descriptor_flags() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        producer.submit_one(0x1000, 256, false).unwrap();

        let desc = ring.read_desc(0);

        // Available descriptor: AVAIL != USED
        let flags = desc.flags();
        assert_ne!(
            flags.contains(DescFlags::AVAIL),
            flags.contains(DescFlags::USED)
        );

        // ... and AVAIL=true, USED=false for wrap=true
        assert!(flags.contains(DescFlags::AVAIL));
        assert!(!flags.contains(DescFlags::USED));
    }

    #[test]
    fn test_used_descriptor_flags() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        producer.submit_one(0x1000, 256, false).unwrap();

        let mut desc = ring.read_desc(0);
        desc.mark_used(true);
        ring.write_desc(0, desc);

        let desc = ring.read_desc(0);
        let flags = desc.flags();

        // Used descriptor: AVAIL == USED
        assert_eq!(
            flags.contains(DescFlags::AVAIL),
            flags.contains(DescFlags::USED)
        );
    }

    #[test]
    fn test_poll_empty_ring() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        // Poll without any submitted buffers
        assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));
    }

    #[test]
    fn test_submit_when_full() {
        let ring = make_ring(2);
        let mut producer = make_producer(&ring);

        producer.submit_one(0x1000, 256, false).unwrap();
        producer.submit_one(0x2000, 256, false).unwrap();

        // Ring is full
        assert!(matches!(
            producer.submit_one(0x3000, 256, false),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn test_wrap_stress() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Do multiple full laps
        for lap in 0..3 {
            let expected_wrap = lap % 2 == 0;

            for _ in 0..4 {
                let id = producer.submit_one(0x1000, 256, false).unwrap();

                let (dev_id, _) = consumer.poll_available().unwrap();
                assert_eq!(dev_id, id);

                consumer.submit_used(dev_id, 256).unwrap();

                producer.poll_used().unwrap();
            }

            // After full lap, wrap should toggle
            assert_eq!(producer.avail_cursor.wrap(), !expected_wrap);
        }
        assert_invariants(&ring, &producer);
    }

    #[test]
    fn test_next_flag_termination() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .readable(0x2000, 256)
            .readable(0x3000, 256)
            .build()
            .unwrap();

        producer.submit_available(&chain).unwrap();

        // First two should have NEXT
        for i in 0..2 {
            let desc = ring.read_desc(i);
            assert!(desc.flags().contains(DescFlags::NEXT));
        }

        // Last should not have NEXT
        let tail_desc = ring.read_desc(2);
        assert!(!tail_desc.flags().contains(DescFlags::NEXT));
    }

    #[test]
    fn test_consumer_initialization() {
        let ring = make_ring(8);
        let consumer = make_consumer(&ring);

        assert_eq!(consumer.avail_cursor.head(), 0);
        assert!(consumer.avail_cursor.wrap());
        assert_eq!(consumer.used_cursor.head(), 0);
        assert!(consumer.used_cursor.wrap());

        for i in 0..8 {
            assert_eq!(consumer.id_num[i], 0);
        }
    }

    #[test]
    fn test_consumer_poll_available_single() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let id = producer.submit_one(0x1000, 512, false).unwrap();

        let (polled_id, chain) = consumer.poll_available().unwrap();

        assert_eq!(polled_id, id);
        assert_eq!(chain.len(), 1);
        assert_eq!(chain.elems()[0].addr, 0x1000);
        assert_eq!(chain.elems()[0].len, 512);
        assert!(!chain.elems()[0].writable);

        // Chain length recorded
        assert_eq!(consumer.id_num[id as usize], 1);
        assert_eq!(consumer.avail_cursor.head(), 1);
    }

    #[test]
    fn test_consumer_poll_available_chain() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .readable(0x2000, 256)
            .writable(0x3000, 512)
            .build()
            .unwrap();

        let id = producer.submit_available(&chain).unwrap();

        let (polled_id, polled_chain) = consumer.poll_available().unwrap();

        assert_eq!(polled_id, id);
        assert_eq!(polled_chain.len(), 3);

        assert_eq!(polled_chain.elems()[0].addr, 0x1000);
        assert!(!polled_chain.elems()[0].writable);

        assert_eq!(polled_chain.elems()[1].addr, 0x2000);
        assert!(!polled_chain.elems()[1].writable);

        assert_eq!(polled_chain.elems()[2].addr, 0x3000);
        assert!(polled_chain.elems()[2].writable);

        assert_eq!(consumer.id_num[id as usize], 3);
    }

    #[test]
    fn test_consumer_rejects_duplicate_inflight_id() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let id = producer.submit_one(0x1000, 512, false).unwrap();
        let (polled_id, _) = consumer.poll_available().unwrap();
        assert_eq!(polled_id, id);

        let mut desc = Descriptor::new(0x2000, 256, id, DescFlags::empty());
        desc.mark_avail(consumer.avail_cursor.wrap());
        ring.write_desc(consumer.avail_cursor.head(), desc);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::InvalidState)
        ));
    }

    #[test]
    fn test_consumer_submit_used() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let id = producer.submit_one(0x1000, 512, true).unwrap();

        let (polled_id, _) = consumer.poll_available().unwrap();

        // Submit as used
        consumer.submit_used(polled_id, 256).unwrap();

        // Check descriptor marked used
        let desc = ring.read_desc(0);

        assert_eq!(desc.id, id);
        assert_eq!(desc.len, 256);
        assert!(desc.is_used(true));

        // Cursor advanced, chain length cleared
        assert_eq!(consumer.used_cursor.head(), 1);
        assert_eq!(consumer.id_num[id as usize], 0);
    }

    #[test]
    fn test_consumer_submit_used_multi_descriptor() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .writable(0x2000, 512)
            .writable(0x3000, 512)
            .build()
            .unwrap();

        producer.submit_available(&chain).unwrap();

        let (id, _) = consumer.poll_available().unwrap();

        consumer.submit_used(id, 1024).unwrap();

        // Only head marked used
        let head_desc = ring.read_desc(0);
        assert!(head_desc.is_used(true));
        assert_eq!(head_desc.len, 1024);

        // Cursor skipped entire chain
        assert_eq!(consumer.used_cursor.head(), 3);
        assert_eq!(consumer.id_num[id as usize], 0);
    }

    #[test]
    fn test_consumer_poll_empty() {
        let ring = make_ring(4);
        let mut consumer = make_consumer(&ring);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn test_consumer_peek() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let consumer = make_consumer(&ring);

        producer.submit_one(0x1000, 512, false).unwrap();
        assert!(consumer.peek_available().unwrap());

        let empty_ring = make_ring(4);
        let empty_consumer = make_consumer(&empty_ring);
        assert!(!empty_consumer.peek_available().unwrap());
    }

    #[test]
    fn test_full_roundtrip() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 256)
            .writable(0x2000, 512)
            .build()
            .unwrap();

        let id = producer.submit_available(&chain).unwrap();

        let (consumer_id, consumer_chain) = consumer.poll_available().unwrap();

        assert_eq!(consumer_id, id);
        assert_eq!(consumer_chain.len(), 2);

        consumer.submit_used(consumer_id, 512).unwrap();

        let used = producer.poll_used().unwrap();
        assert_eq!(used.id, id);
        assert_eq!(used.len, 512);
    }

    #[test]
    fn ring_initial_poll_used_blocks() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        // No submissions yet: all descriptors zero.
        for _ in 0..8 {
            assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));
        }
        // Invariants: num_free == ring size
        assert_eq!(producer.num_free, ring.len());
    }

    #[test]
    fn ring_consumer_blocks_until_submit() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 32)
            .readable(0x2000, 16)
            .build()
            .unwrap();

        let id = producer.submit_available(&chain).unwrap();

        let (cid, polled) = consumer.poll_available().unwrap();
        assert_eq!(cid, id);
        assert_eq!(polled.len(), chain.len());
    }

    #[test]
    fn test_out_of_order_completion_stream() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Driver submits two single-descriptor chains A then B
        let id_a = producer.submit_one(0x1000, 256, true).unwrap();
        let id_b = producer.submit_one(0x2000, 256, true).unwrap();

        // Device polls them in ring order (A then B)
        let (dev_id_a, chain_a) = consumer.poll_available().unwrap();
        assert_eq!(dev_id_a, id_a);
        assert_eq!(chain_a.len(), 1);

        let (dev_id_b, chain_b) = consumer.poll_available().unwrap();
        assert_eq!(dev_id_b, id_b);
        assert_eq!(chain_b.len(), 1);

        // Device completes B first, then A
        consumer.submit_used(dev_id_b, 128).unwrap();
        consumer.submit_used(dev_id_a, 256).unwrap();

        // Driver polls used stream: should see B (first completion)
        let used_b = producer.poll_used().unwrap();
        assert_eq!(used_b.id, id_b);
        assert_eq!(used_b.len, 128);

        // Then sees A
        let used_a = producer.poll_used().unwrap();
        assert_eq!(used_a.id, id_a);
        assert_eq!(used_a.len, 256);

        // IDs recycled
        assert!(producer.id_free.contains(&id_a));
        assert!(producer.id_free.contains(&id_b));
    }

    #[test]
    fn test_mixed_chain_sizes_out_of_order_completion() {
        let ring = make_ring(16);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let chains = vec![
            BufferChainBuilder::new()
                .readable(0x1000, 10)
                .writable(0x2000, 5)
                .build()
                .unwrap(),
            BufferChainBuilder::new()
                .readable(0x3000, 8)
                .readable(0x3010, 8)
                .writable(0x3020, 16)
                .build()
                .unwrap(),
            BufferChainBuilder::new()
                .readable(0x4000, 4)
                .build()
                .unwrap(),
            BufferChainBuilder::new()
                .readable(0x5000, 4)
                .readable(0x5010, 4)
                .readable(0x5020, 4)
                .writable(0x5030, 4)
                .build()
                .unwrap(),
        ];

        for c in &chains {
            producer.submit_available(c).unwrap();
        }

        let mut dev_chain_lens = Vec::new();
        for _ in &chains {
            let (id, chain) = consumer.poll_available().unwrap();
            dev_chain_lens.push((id, chain.len() as u32));
        }

        let order = [1, 3, 0, 2];
        let mut completion = Vec::new();

        for &idx in &order {
            let (id, len) = dev_chain_lens[idx];
            consumer.submit_used(id, len).unwrap();
            completion.push((id, len));
        }

        for (expected_id, expected_len) in &completion {
            let used = producer.poll_used().unwrap();
            assert_eq!(used.id, *expected_id);
            assert_eq!(used.len, *expected_len);
            assert_eq!(producer.id_num[*expected_id as usize], 0);
            assert!(producer.id_free.contains(expected_id));
        }

        assert_invariants(&ring, &producer);
    }

    // Used stream wrap crossing
    #[test]
    fn test_used_stream_wrap_crossing() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Submit enough single descriptors to make used writes wrap
        let mut ids = Vec::new();
        for i in 0..8 {
            ids.push(producer.submit_one(0x1000 + i as u64, 1, false).unwrap());
        }

        // Device polls all
        for _ in 0..8 {
            consumer.poll_available().unwrap();
        }

        // Complete all in order except we simulate out-of-order by reversing
        for &id in ids.iter().rev() {
            consumer.submit_used(id, 1).unwrap();
        }

        // Producer polls used; after consuming size descriptors used_cursor should wrap
        for _ in 0..8 {
            producer.poll_used().unwrap();
        }
        assert_eq!(producer.used_cursor.head(), 0);
        assert!(!producer.used_cursor.wrap()); // flipped once
        assert_invariants(&ring, &producer);
    }

    // Interleaved availability and completion
    #[test]
    fn test_interleaved_submit_completion() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Submit chain A (len 2)
        let chain_a = BufferChainBuilder::new()
            .readable(0x1000, 8)
            .writable(0x2000, 8)
            .build()
            .unwrap();
        let id_a = producer.submit_available(&chain_a).unwrap();

        // Device polls A
        let (dev_id_a, _) = consumer.poll_available().unwrap();
        assert_eq!(dev_id_a, id_a);

        // Device completes A
        consumer.submit_used(dev_id_a, 8).unwrap();

        // Submit chain B (len 3) before driver reclaims A
        let chain_b = BufferChainBuilder::new()
            .readable(0x3000, 4)
            .readable(0x3010, 4)
            .writable(0x3020, 4)
            .build()
            .unwrap();
        let id_b = producer.submit_available(&chain_b).unwrap();

        // Device polls B
        let (dev_id_b, _) = consumer.poll_available().unwrap();
        assert_eq!(dev_id_b, id_b);

        // Driver reclaims A
        let used_a = producer.poll_used().unwrap();
        assert_eq!(used_a.id, id_a);

        // Device completes B
        consumer.submit_used(dev_id_b, 12).unwrap();

        // Driver reclaims B
        let used_b = producer.poll_used().unwrap();
        assert_eq!(used_b.id, id_b);

        assert_invariants(&ring, &producer);
    }

    // Partial publish safety (head not published yet)
    #[test]
    fn test_partial_publish_safety() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);
        let mut producer = make_producer(&ring);

        // Build chain manually: write tails only
        let chain = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .readable(0x2000, 4)
            .writable(0x3000, 4)
            .build()
            .unwrap();

        // Simulate manual tail writes without head publish
        let id = producer.id_free.pop().unwrap();
        producer.id_num[id as usize] = chain.len() as u16;

        // Emulate internal position logic
        let head_idx = producer.avail_cursor.head();
        let wrap_start = producer.avail_cursor.wrap();
        let mut pos = producer.avail_cursor;
        pos.advance();

        for (i, elem) in chain.elems().iter().enumerate().skip(1) {
            let is_next = i + 1 < chain.len();
            let mut flags = DescFlags::empty();
            flags.set(DescFlags::NEXT, is_next);
            flags.set(DescFlags::WRITE, elem.writable);
            let mut d = Descriptor::new(elem.addr, elem.len, id, flags);
            d.mark_avail(pos.wrap());
            ring.write_desc(pos.head(), d);
            pos.advance();
        }

        // Head not published yet: consumer must not see chain
        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));

        // Now publish head
        let head_elem = chain.elems()[0];
        let mut head_flags = DescFlags::empty();
        head_flags.set(DescFlags::NEXT, true);
        head_flags.set(DescFlags::WRITE, head_elem.writable);
        let mut head_desc = Descriptor::new(head_elem.addr, head_elem.len, id, head_flags);
        head_desc.mark_avail(wrap_start);
        ring.write_desc(head_idx, head_desc);
        producer.avail_cursor = pos;
        producer.num_free -= chain.len();

        // Consumer can now see the chain
        let (dev_id, dev_chain) = consumer.poll_available().unwrap();
        assert_eq!(dev_id, id);
        assert_eq!(dev_chain.len(), chain.len());
        assert_invariants(&ring, &producer);
    }

    // Tail misuse negative test
    #[test]
    fn test_tail_marked_used_ignored() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .readable(0x2000, 4)
            .build()
            .unwrap();
        let id = producer.submit_available(&chain).unwrap();

        // Incorrectly mark tail (index 1) used
        let mut tail_desc = ring.read_desc(1);
        tail_desc.mark_used(producer.used_cursor.wrap());
        ring.write_desc(1, tail_desc);

        // Poll should return WouldBlock (head not used yet)
        assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));

        // Mark head used properly
        let mut head_desc = ring.read_desc(0);
        head_desc.mark_used(producer.used_cursor.wrap());
        ring.write_desc(0, head_desc);

        // Now poll succeeds
        let used = producer.poll_used().unwrap();
        assert_eq!(used.id, id);
        assert_invariants(&ring, &producer);
    }

    // Max chain length boundary
    #[test]
    fn test_max_chain_len_rejected() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        // Try chain longer than ring size
        let elems = (0..9).map(|i| BufferElement {
            addr: 0x1000 + i as u64,
            len: 42,
            writable: false,
        });

        let chain = BufferChainBuilder::new().readables(elems).build().unwrap();

        // Submit_available should reject when num_free < total_descs
        assert!(matches!(
            producer.submit_available(&chain),
            Err(RingError::WouldBlock)
        ));
    }

    // Descriptor state monotonicity after many cycles
    #[test]
    fn test_descriptor_state_monotonicity() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Track states: 0=zero/init, 1=available, 2=used, 3=reclaimed
        let mut states = vec![0u8; 8];

        for _ in 0..5 {
            for state in states.iter_mut() {
                let id = producer.submit_one(0x1000, 4, false).unwrap();
                // mark available
                *state = (*state).max(1);

                // device polls and completes
                let (dev_id, _) = consumer.poll_available().unwrap();
                consumer.submit_used(dev_id, 4).unwrap();
                *state = (*state).max(2);

                let used = producer.poll_used().unwrap();
                assert_eq!(used.id, id);
                *state = (*state).max(3);
            }

            assert_invariants(&ring, &producer);
        }

        // Ensure monotonic progression (never decrease)
        for s in states {
            assert!(s >= 3);
        }
    }

    // Large multi-lap random submission/completion
    #[test]
    fn test_random_stress_small() {
        use rand::RngExt;
        use rand::seq::SliceRandom;

        let ring = make_ring(16);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);
        let mut rng = rand::rng();

        // Submit initial set
        let mut active_ids = Vec::new();
        for _ in 0..8 {
            let len = rng.random_range(1..=4);
            let mut b = BufferChainBuilder::new().readable(0x1000, 4);
            for i in 1..len {
                b = b.readable(0x1000 + i as u64 * 0x10, 4);
            }
            let chain = b.build().unwrap();
            if let Ok(id) = producer.submit_available(&chain) {
                active_ids.push(id);
            }
        }

        let mut dev_ids = Vec::new();
        while let Ok((id, _)) = consumer.poll_available() {
            dev_ids.push(id);
        }

        // Randomly complete
        dev_ids.shuffle(&mut rng);
        for id in &dev_ids {
            let chain_len = consumer.id_num[*id as usize];
            consumer.submit_used(*id, chain_len as u32 * 4).unwrap();
        }
        // Driver reclaim
        for _ in &dev_ids {
            if producer.poll_used().is_ok() {}
        }

        assert_invariants(&ring, &producer);
    }

    // Out-of-order multi-length explicit
    #[test]
    fn test_out_of_order_multi_length() {
        let ring = make_ring(16);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let chain_a = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .writable(0x2000, 4)
            .build()
            .unwrap();
        let chain_b = BufferChainBuilder::new()
            .readable(0x3000, 4)
            .readable(0x3010, 4)
            .writable(0x3020, 4)
            .build()
            .unwrap();
        let chain_c = BufferChainBuilder::new()
            .readable(0x4000, 4)
            .build()
            .unwrap();

        let id_a = producer.submit_available(&chain_a).unwrap();
        let id_b = producer.submit_available(&chain_b).unwrap();
        let id_c = producer.submit_available(&chain_c).unwrap();

        let (d_a, _) = consumer.poll_available().unwrap();
        let (d_b, _) = consumer.poll_available().unwrap();
        let (d_c, _) = consumer.poll_available().unwrap();
        assert_eq!(d_a, id_a);
        assert_eq!(d_b, id_b);
        assert_eq!(d_c, id_c);

        // Complete B, then C, then A
        consumer.submit_used(d_b, 12).unwrap();
        consumer.submit_used(d_c, 4).unwrap();
        consumer.submit_used(d_a, 8).unwrap();

        let u_b = producer.poll_used().unwrap();
        assert_eq!(u_b.id, id_b);
        let u_c = producer.poll_used().unwrap();
        assert_eq!(u_c.id, id_c);
        let u_a = producer.poll_used().unwrap();
        assert_eq!(u_a.id, id_a);

        assert_invariants(&ring, &producer);
    }

    #[test]
    fn interleave_submit_and_completion() {
        let ring = make_ring(16);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Submit A (len 2)
        let chain_a = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .writable(0x2000, 4)
            .build()
            .unwrap();
        let id_a = producer.submit_available(&chain_a).unwrap();

        // Device polls A
        let (d_a, _) = consumer.poll_available().unwrap();
        assert_eq!(d_a, id_a);

        // Immediately complete A
        consumer.submit_used(d_a, 8).unwrap();

        // Submit B (len 3)
        let chain_b = BufferChainBuilder::new()
            .readable(0x3000, 4)
            .readable(0x3010, 4)
            .writable(0x3020, 4)
            .build()
            .unwrap();
        let id_b = producer.submit_available(&chain_b).unwrap();

        // Driver polls used: gets A
        let u_a = producer.poll_used().unwrap();
        assert_eq!(u_a.id, id_a);
        assert_eq!(u_a.len, 8);

        // Device polls B and submits used for it
        let (d_b, _) = consumer.poll_available().unwrap();
        assert_eq!(d_b, id_b);
        consumer.submit_used(d_b, 12).unwrap();

        // Submit C (len 1)
        let id_c = producer.submit_one(0x4000, 4, false).unwrap();

        // Device polls C and completes it
        let (d_c, _) = consumer.poll_available().unwrap();
        assert_eq!(d_c, id_c);
        consumer.submit_used(d_c, 4).unwrap();

        // Driver polls used: gets B then C
        let u_b = producer.poll_used().unwrap();
        assert_eq!(u_b.id, id_b);
        assert_eq!(u_b.len, 12);

        let u_c = producer.poll_used().unwrap();
        assert_eq!(u_c.id, id_c);
        assert_eq!(u_c.len, 4);

        assert_invariants(&ring, &producer);
    }

    // Event suppression tests
    #[test]
    fn producer_disable_used_notifications_writes_driver_disable() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        assert_eq!(ring.read_driver_event().flags(), EventFlags::ENABLE);
        producer.disable_used_notifications().unwrap();
        assert_eq!(ring.read_driver_event().flags(), EventFlags::DISABLE);
    }

    #[test]
    fn producer_enable_used_notifications_writes_driver_enable() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        producer.disable_used_notifications().unwrap();
        assert_eq!(ring.read_driver_event().flags(), EventFlags::DISABLE);

        producer.enable_used_notifications().unwrap();
        assert_eq!(ring.read_driver_event().flags(), EventFlags::ENABLE);
    }

    #[test]
    fn producer_enable_used_notifications_desc_sets_off_wrap_and_flags() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        producer.enable_used_notifications_desc(5, true).unwrap();

        let evt = ring.read_driver_event();
        assert_eq!(evt.flags(), EventFlags::DESC);
        assert_eq!(evt.desc_event_off(), 5);
        assert!(evt.desc_event_wrap());
    }

    #[test]
    fn producer_enable_used_notifications_for_next_programs_used_cursor() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        // initial used cursor: head=0, wrap=true
        producer.enable_used_notifications_for_next().unwrap();

        let evt = ring.read_driver_event();
        assert_eq!(evt.flags(), EventFlags::DESC);
        assert_eq!(evt.desc_event_off(), 0);
        assert!(evt.desc_event_wrap());
    }

    #[test]
    fn consumer_disable_avail_notifications_writes_device_disable() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        assert_eq!(ring.read_device_event().flags(), EventFlags::ENABLE);
        consumer.disable_avail_notifications().unwrap();
        assert_eq!(ring.read_device_event().flags(), EventFlags::DISABLE);
    }

    #[test]
    fn consumer_enable_avail_notifications_writes_device_enable() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        consumer.disable_avail_notifications().unwrap();
        assert_eq!(ring.read_device_event().flags(), EventFlags::DISABLE);

        consumer.enable_avail_notifications().unwrap();
        assert_eq!(ring.read_device_event().flags(), EventFlags::ENABLE);
    }

    #[test]
    fn consumer_enable_avail_notifications_desc_sets_off_wrap_and_flags() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        consumer.enable_avail_notifications_desc(7, false).unwrap();

        let evt = ring.read_device_event();
        assert_eq!(evt.flags(), EventFlags::DESC);
        assert_eq!(evt.desc_event_off(), 7);
        assert!(!evt.desc_event_wrap());
    }

    #[test]
    fn consumer_enable_avail_notifications_for_next_programs_avail_cursor() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        // initial avail cursor: head=0, wrap=true
        consumer.enable_avail_notifications_for_next().unwrap();

        let evt = ring.read_device_event();
        assert_eq!(evt.flags(), EventFlags::DESC);
        assert_eq!(evt.desc_event_off(), 0);
        assert!(evt.desc_event_wrap());
    }

    #[test]
    fn producer_does_not_write_device_event_when_toggling_used_notifications() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let dev_before = ring.read_device_event();
        producer.disable_used_notifications().unwrap();
        let dev_after = ring.read_device_event();

        assert_eq!(dev_after, dev_before);
    }

    #[test]
    fn consumer_does_not_write_driver_event_when_toggling_avail_notifications() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let drv_before = ring.read_driver_event();
        consumer.disable_avail_notifications().unwrap();
        let drv_after = ring.read_driver_event();

        assert_eq!(drv_after, drv_before);
    }

    #[test]
    fn should_notify_flags_enable_disable() {
        let ring_len = 8;

        let old = RingCursor {
            head: 0,
            size: ring_len,
            wrap: true,
        };
        let new = RingCursor {
            head: 1,
            size: ring_len,
            wrap: true,
        };

        // DISABLE -> never notify
        let evt = EventSuppression::new(0, EventFlags::DISABLE);
        assert!(!should_notify(evt, ring_len, old, new));

        // ENABLE -> always notify
        let evt = EventSuppression::new(0, EventFlags::ENABLE);
        assert!(should_notify(evt, ring_len, old, new));
    }

    #[test]
    fn should_notify_desc_no_crossing() {
        let ring_len = 8;

        let old = RingCursor {
            head: 2,
            size: ring_len,
            wrap: true,
        };
        let new = RingCursor {
            head: 3,
            size: ring_len,
            wrap: true,
        };

        // event at 6, we did not cross it
        let mut evt = EventSuppression::zeroed();
        evt.set_desc_event(6, true);
        evt.set_flags(EventFlags::DESC);

        assert!(!should_notify(evt, ring_len, old, new));
    }

    #[test]
    fn should_notify_desc_wrap_mismatch_adjusts_event_idx() {
        let ring_len = 8;

        let old = RingCursor {
            head: 7,
            size: ring_len,
            wrap: true,
        };
        let new = RingCursor {
            head: 1,
            size: ring_len,
            wrap: false,
        };

        let mut evt = EventSuppression::zeroed();
        evt.set_desc_event(7, true);
        evt.set_flags(EventFlags::DESC);

        assert!(should_notify(evt, ring_len, old, new));
    }

    #[test]
    fn ring_need_event_basic_cases() {
        // If event_idx == new-1, should be true
        assert!(ring_need_event(4, 5, 2));
        // If no progress, should be false
        assert!(!ring_need_event(4, 5, 5));

        // Wrapping arithmetic sanity: old near u16::MAX
        let old = 0xFFFE;
        let new = 1;
        // event at 0xFFFF is considered "just before wrap"
        assert!(ring_need_event(0xFFFF, new, old));
    }

    // Bad device/driver tests
    #[test]
    fn bad_device_marks_tail_used() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .readable(0x2000, 4)
            .build()
            .unwrap();
        let id = producer.submit_available(&chain).unwrap();

        // Bad device: mark index 1 (tail) used
        let mut tail = ring.read_desc(1);
        tail.mark_used(producer.used_cursor.wrap());
        ring.write_desc(1, tail);

        // Driver must not consume it
        assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));

        // Now mark head properly, driver must consume
        let mut head = ring.read_desc(0);
        head.mark_used(producer.used_cursor.wrap());
        ring.write_desc(0, head);

        let used = producer.poll_used().unwrap();
        assert_eq!(used.id, id);
    }

    #[test]
    fn bad_device_wrong_used_bits() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        let id = producer.submit_one(0x1000, 8, true).unwrap();

        // Malformed: set AVAIL but clear USED (should be equal for used)
        let mut d = ring.read_desc(0);
        // Force flags to look like "available" despite intent
        d.mark_avail(producer.used_cursor.wrap());
        d.len = 8;
        ring.write_desc(0, d);

        assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));

        let mut d2 = ring.read_desc(0);
        d2.mark_used(producer.used_cursor.wrap());
        ring.write_desc(0, d2);

        let u = producer.poll_used().unwrap();
        assert_eq!(u.id, id);
    }

    #[test]
    fn bad_driver_next_never_clears() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);
        let mut producer = make_producer(&ring);

        // Allocate an ID and pretend one huge chain
        let id = producer.id_free.pop().unwrap();
        producer.id_num[id as usize] = 8;

        let mut pos = producer.avail_cursor;
        let wrap_start = pos.wrap();

        // Write every descriptor with NEXT set and same id
        for _ in 0..8 {
            let idx = pos.head();
            let mut flags = DescFlags::empty();
            flags.set(DescFlags::NEXT, true); // incorrect: last should NOT have NEXT
            let mut desc = Descriptor::new(0x1000 + idx as u64 * 0x10, 4, id, flags);
            desc.mark_avail(pos.wrap());
            ring.write_desc(idx, desc);
            pos.advance();
        }

        // Publish head last (simulate driver behavior)
        let head_idx = producer.avail_cursor.head();
        let mut head_flags = DescFlags::empty();
        head_flags.set(DescFlags::NEXT, true);
        let mut head_desc = Descriptor::new(0x42, 4, id, head_flags);
        head_desc.mark_avail(wrap_start);
        ring.write_desc(head_idx, head_desc);

        // Consumer should detect invalid chain via step guard
        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn bad_driver_interleaved_readables_and_writables() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .readable(0x2000, 4)
            .writable(0x2000, 4)
            .build()
            .unwrap();

        let _id = producer.submit_available(&chain).unwrap();

        // now change first descriptor to writable (bad driver)
        let mut first = ring.read_desc(0);
        first.flags |= DescFlags::WRITE.bits();
        ring.write_desc(0, first);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn bad_device_marks_multiple_used_in_chain() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let chain = BufferChainBuilder::new()
            .readable(0x1000, 4)
            .readable(0x2000, 4)
            .build()
            .unwrap();
        let id = producer.submit_available(&chain).unwrap();

        // Bad device: mark head and tail used
        let mut head = ring.read_desc(0);
        head.mark_used(producer.used_cursor.wrap());
        ring.write_desc(0, head);

        let mut tail = ring.read_desc(1);
        tail.mark_used(producer.used_cursor.wrap());
        ring.write_desc(1, tail);

        // Driver consumes once
        let u = producer.poll_used().unwrap();
        assert_eq!(u.id, id);

        // Next poll should block; no duplicate consumption
        assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));
    }

    #[test]
    fn bad_device_writes_used_at_wrong_slot() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let _id = producer.submit_one(0x1000, 4, true).unwrap();

        // Wrong slot: mark index 3 used while next_used is 0
        let mut d = ring.read_desc(3);
        d.mark_used(producer.used_cursor.wrap());
        ring.write_desc(3, d);

        // Driver should still block (polls only slot 0)
        assert!(matches!(producer.poll_used(), Err(RingError::WouldBlock)));

        // Now mark slot 0 correctly, driver can consume
        let mut d0 = ring.read_desc(0);
        d0.mark_used(producer.used_cursor.wrap());
        ring.write_desc(0, d0);
        let _u = producer.poll_used().unwrap();
    }

    #[test]
    fn bad_driver_reuses_id_while_outstanding() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        // Submit first buffer: allocate ID
        let id = producer.submit_one(0x1000, 4, false).unwrap();
        assert_eq!(producer.id_num[id as usize], 1);

        // push the same ID back into free list while it's still outstanding.
        producer.id_free.push(id);

        // Next submit should fail because ID is still outstanding.
        let res = producer.submit_one(0x2000, 4, false);
        assert!(matches!(res, Err(RingError::InvalidState)));
    }

    #[test]
    fn test_avail_cursor_accessor() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        // Initial cursor
        let cursor = producer.avail_cursor();
        assert_eq!(cursor.head(), 0);
        assert!(cursor.wrap());

        // After submit
        producer.submit_one(0x1000, 512, false).unwrap();
        let cursor = producer.avail_cursor();
        assert_eq!(cursor.head(), 1);
        assert!(cursor.wrap());
    }

    #[test]
    fn test_should_notify_since() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        let before = producer.avail_cursor();
        producer.submit_one(0x1000, 512, false).unwrap();

        // Default is ENABLE mode, so should notify
        let should_notify = producer.should_notify_since(before).unwrap();
        assert!(should_notify);
    }

    #[test]
    fn test_batch_notification_single_check() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let before = producer.avail_cursor();

        // Submit multiple descriptors
        producer.submit_one(0x1000, 512, false).unwrap();
        producer.submit_one(0x2000, 512, false).unwrap();
        producer.submit_one(0x3000, 512, false).unwrap();

        // Single notification check for the entire batch
        let should_notify = producer.should_notify_since(before).unwrap();
        assert!(should_notify);

        // Consumer sees all 3 descriptors
        for _ in 0..3 {
            let (_, _) = consumer.poll_available().unwrap();
        }
    }

    #[test]
    fn test_ring_cursor_reset() {
        let mut cursor = RingCursor::new(16);
        cursor.advance_by(5);
        assert_eq!(cursor.head(), 5);

        cursor.reset();
        assert_eq!(cursor, RingCursor::new(16));
        assert_eq!(cursor.head(), 0);
        assert!(cursor.wrap());
    }

    #[test]
    fn test_ring_cursor_reset_after_wrap() {
        let mut cursor = RingCursor::new(4);
        // Advance past the wrap point
        cursor.advance_by(5);
        assert_eq!(cursor.head(), 1);
        assert!(!cursor.wrap());

        cursor.reset();
        assert_eq!(cursor.head(), 0);
        assert!(cursor.wrap());
    }

    #[test]
    fn test_ring_producer_reset_matches_new() {
        let ring = make_ring(8);
        let fresh = make_producer(&ring);

        let mut used = make_producer(&ring);
        // Mutate state
        used.submit_one(0x1000, 64, false).unwrap();
        used.submit_one(0x2000, 128, true).unwrap();

        used.reset();

        assert_eq!(used.avail_cursor, fresh.avail_cursor);
        assert_eq!(used.used_cursor, fresh.used_cursor);
        assert_eq!(used.num_free, fresh.num_free);
        assert_eq!(used.id_free.len(), fresh.id_free.len());
        assert_eq!(used.id_num.as_slice(), fresh.id_num.as_slice());
        assert_eq!(used.event_flags_shadow, fresh.event_flags_shadow);
    }

    #[test]
    fn test_ring_producer_reset_id_free_complete() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        // Submit and consume several descriptors
        for i in 0..4u64 {
            producer.submit_one(0x1000 + i * 0x100, 64, false).unwrap();
        }
        assert_eq!(producer.num_free, 4);

        producer.reset();

        assert_eq!(producer.num_free, 8);
        assert_eq!(producer.id_free.len(), 8);
        // All IDs 0..8 should be present
        for id in 0..8u16 {
            assert!(producer.id_free.contains(&id));
        }
    }

    #[test]
    fn test_ring_consumer_reset_matches_new() {
        let ring = make_ring(8);
        let fresh = make_consumer(&ring);

        let mut used = make_consumer(&ring);

        // Submit from producer side so consumer has something to poll
        let mut producer = make_producer(&ring);
        producer.submit_one(0x1000, 64, false).unwrap();

        // Consumer polls the available descriptor
        let (id, _chain) = used.poll_available().unwrap();
        used.submit_used(id, 64).unwrap();

        used.reset();

        assert_eq!(used.avail_cursor, fresh.avail_cursor);
        assert_eq!(used.used_cursor, fresh.used_cursor);
        assert_eq!(used.id_num.as_slice(), fresh.id_num.as_slice());
        assert_eq!(used.num_inflight, fresh.num_inflight);
        assert_eq!(used.event_flags_shadow, fresh.event_flags_shadow);
    }

    #[test]
    fn test_ring_consumer_reset_clears_inflight() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Submit and poll two items (consume but do not complete)
        producer.submit_one(0x1000, 64, false).unwrap();
        producer.submit_one(0x2000, 64, false).unwrap();
        let _ = consumer.poll_available().unwrap();
        let _ = consumer.poll_available().unwrap();
        assert_eq!(consumer.num_inflight, 2);

        consumer.reset();
        assert_eq!(consumer.num_inflight, 0);
    }

    #[test]
    fn test_reset_prefilled_sets_cursors() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let ids: Vec<u16> = (0..8).collect();
        producer.reset_prefilled(&ids);

        // avail wrapped once (all 8 slots submitted)
        assert_eq!(producer.avail_cursor.head(), 0);
        assert!(!producer.avail_cursor.wrap());
        // used cursor at initial position
        assert_eq!(producer.used_cursor.head(), 0);
        assert!(producer.used_cursor.wrap());
    }

    #[test]
    fn test_reset_prefilled_all_ids_inflight() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let ids: Vec<u16> = (0..8).collect();
        producer.reset_prefilled(&ids);

        assert_eq!(producer.num_free, 0);
        assert!(producer.id_free.is_empty());
        assert!(producer.id_num.iter().all(|&n| n == 1));
    }

    #[test]
    fn test_reset_prefilled_partial() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        producer.reset_prefilled(&[5, 6, 7, 3]);

        // avail cursor at position 4, no wrap
        assert_eq!(producer.avail_cursor.head(), 4);
        assert!(producer.avail_cursor.wrap());
        // used cursor at initial position
        assert_eq!(producer.used_cursor.head(), 0);
        assert!(producer.used_cursor.wrap());

        assert_eq!(producer.num_free, 4);
        assert_eq!(producer.id_free.len(), 4);
        for &id in &[0, 1, 2, 4] {
            assert!(producer.id_free.contains(&id));
        }
        // Only the specified IDs are in-flight
        for &id in &[5, 6, 7, 3] {
            assert_eq!(producer.id_num[id as usize], 1);
        }
        for &id in &[0, 1, 2, 4] {
            assert_eq!(producer.id_num[id as usize], 0);
        }
    }

    #[test]
    fn test_reset_prefilled_partial_then_submit() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        producer.reset_prefilled(&[4, 5, 6, 7]);

        let id = producer.submit_one(0x8000, 128, false).unwrap();

        assert!([0, 1, 2, 3].contains(&id));
        assert_eq!(producer.num_free, 3);
        assert_eq!(producer.id_num[id as usize], 1);
    }

    #[test]
    fn test_reset_prefilled_then_poll_used() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        // Simulate host prefill: LIFO assigns IDs 3, 2, 1, 0
        for i in 0..4u64 {
            producer.submit_one(0x1000 + i * 4096, 4096, true).unwrap();
        }

        // Consumer marks one as used
        let mut consumer = make_consumer(&ring);
        let (id, _chain) = consumer.poll_available().unwrap();
        consumer.submit_used(id, 64).unwrap();

        // Fresh producer restores via reset_prefilled with all IDs
        let mut restored = make_producer(&ring);
        restored.reset_prefilled(&[0, 1, 2, 3]);

        // poll_used should discover the consumed descriptor
        let used = restored.poll_used().unwrap();
        assert_eq!(used.id, id);
    }

    #[test]
    fn test_desc_table_read_after_submit() {
        let ring = make_ring(8);
        let mut writer = make_producer(&ring);
        writer.submit_one(0x1000, 4096, true).unwrap();

        let reader = make_producer(&ring);
        let addr = reader.desc_table().desc_addr(0).unwrap();
        let flags = Descriptor::read_flags_acquire(reader.mem(), addr).unwrap();
        let desc = Descriptor::read_body(reader.mem(), addr, flags).unwrap();
        assert_eq!(desc.addr, 0x1000);
        assert_eq!(desc.len, 4096);
        assert!(desc.is_writable());
        assert!(desc.is_avail(true));
        assert!(!desc.is_used(true));
    }

    #[test]
    fn test_desc_table_out_of_bounds() {
        let ring = make_ring(8);
        let reader = make_producer(&ring);
        assert!(reader.desc_table().desc_addr(8).is_none());
    }

    #[test]
    fn test_desc_table_read_used_descriptor() {
        let ring = make_ring(8);
        let mut writer = make_producer(&ring);
        writer.submit_one(0x1000, 4096, true).unwrap();

        let mut consumer = make_consumer(&ring);
        let (id, _chain) = consumer.poll_available().unwrap();
        consumer.submit_used(id, 128).unwrap();

        let reader = make_producer(&ring);
        let addr = reader.desc_table().desc_addr(0).unwrap();
        let flags = Descriptor::read_flags_acquire(reader.mem(), addr).unwrap();
        let desc = Descriptor::read_body(reader.mem(), addr, flags).unwrap();
        assert!(desc.is_used(true));
        assert!(!desc.is_avail(true));
    }
}

// Adopted from https://github.com/weltling/virtio-villain/tree/main/tests/packed
#[cfg(test)]
mod virtio_villain {
    use super::tests::{OwnedRing, make_consumer, make_producer, make_ring};
    use super::*;

    fn raw_desc(id: u16, len: u32, flags: DescFlags) -> Descriptor {
        Descriptor::new(0x1000 + u64::from(id) * 0x10, len, id, flags)
    }

    fn avail_desc(id: u16, len: u32, flags: DescFlags) -> Descriptor {
        let mut desc = raw_desc(id, len, flags);
        desc.mark_avail(true);
        desc
    }

    fn write_avail(ring: &OwnedRing, idx: u16, id: u16, flags: DescFlags) {
        write_avail_len(ring, idx, id, 16, flags);
    }

    fn write_avail_len(ring: &OwnedRing, idx: u16, id: u16, len: u32, flags: DescFlags) {
        ring.write_desc(idx, avail_desc(id, len, flags));
    }

    fn write_driver_event(ring: &OwnedRing, off_wrap: u16, flags: u16) {
        write_event(ring, ring.layout().drv_evt_addr(), off_wrap, flags);
    }

    fn write_device_event(ring: &OwnedRing, off_wrap: u16, flags: u16) {
        write_event(ring, ring.layout().dev_evt_addr(), off_wrap, flags);
    }

    fn write_event(ring: &OwnedRing, addr: u64, off_wrap: u16, flags: u16) {
        ring.mem()
            .write(
                addr,
                &[
                    (off_wrap & 0xff) as u8,
                    (off_wrap >> 8) as u8,
                    (flags & 0xff) as u8,
                    (flags >> 8) as u8,
                ],
            )
            .unwrap();
    }

    #[test]
    fn p0003_chain_with_next_past_ring_is_rejected() {
        let ring = make_ring(4);
        let mut consumer = make_consumer(&ring);

        for idx in 0..ring.len() as u16 {
            write_avail(&ring, idx, 0, DescFlags::NEXT);
        }

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0005_indirect_without_feature_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::INDIRECT);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0006_indirect_with_bad_flags_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::INDIRECT | DescFlags::WRITE);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0007_indirect_tail_in_mixed_chain_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::NEXT);
        write_avail(&ring, 1, 0, DescFlags::INDIRECT | DescFlags::WRITE);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0008_long_chain_is_consumable_at_ring_layer() {
        let ring = make_ring(16);
        let mut consumer = make_consumer(&ring);

        for idx in 0..12u16 {
            let flags = if idx == 11 {
                DescFlags::WRITE
            } else {
                DescFlags::NEXT | DescFlags::WRITE
            };
            write_avail(&ring, idx, 0, flags);
        }

        let (id, chain) = consumer.poll_available().unwrap();
        assert_eq!(id, 0);
        assert_eq!(chain.len(), 12);
    }

    #[test]
    fn p0009_unaligned_descriptor_address_is_preserved() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let mut desc = Descriptor::new(0x1003, 16, 0, DescFlags::empty());
        desc.mark_avail(true);
        ring.write_desc(0, desc);

        let (_, chain) = consumer.poll_available().unwrap();
        assert_eq!(chain.elems()[0].addr, 0x1003);
    }

    #[test]
    fn p0010_writable_before_readable_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::NEXT);
        write_avail(&ring, 1, 0, DescFlags::NEXT | DescFlags::WRITE);
        write_avail(&ring, 2, 0, DescFlags::NEXT);
        write_avail(&ring, 3, 0, DescFlags::WRITE);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0011_stale_wrap_descriptor_is_ignored() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let mut desc = raw_desc(0, 16, DescFlags::empty());
        desc.mark_avail(false);
        ring.write_desc(0, desc);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0012_multidescriptor_oob_id_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);
        let bad_id = ring.len() as u16;

        write_avail(&ring, 0, bad_id, DescFlags::NEXT);
        write_avail(&ring, 1, bad_id, DescFlags::WRITE);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::InvalidState)
        ));
    }

    #[test]
    fn p0013_duplicate_inflight_buffer_id_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::NEXT);
        write_avail(&ring, 1, 0, DescFlags::WRITE);
        write_avail(&ring, 2, 0, DescFlags::NEXT);
        write_avail(&ring, 3, 0, DescFlags::WRITE);

        let (first_id, first_chain) = consumer.poll_available().unwrap();
        assert_eq!(first_id, 0);
        assert_eq!(first_chain.len(), 2);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::InvalidState)
        ));
    }

    #[test]
    fn p0014_reserved_driver_event_flags_do_not_block_processing() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, 0, 3);
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, chain) = consumer.poll_available().unwrap();
        assert_eq!(id, 0);
        assert_eq!(chain.len(), 1);
    }

    #[test]
    fn p0015_wrap_at_zero_is_consumed_correctly() {
        let ring = make_ring(2);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        producer.submit_one(0x1000, 16, false).unwrap();
        producer.submit_one(0x2000, 16, false).unwrap();

        assert_eq!(producer.avail_cursor().head(), 0);
        assert!(!producer.avail_cursor().wrap());

        assert_eq!(consumer.poll_available().unwrap().1.elems()[0].addr, 0x1000);
        assert_eq!(consumer.poll_available().unwrap().1.elems()[0].addr, 0x2000);
        assert_eq!(consumer.avail_cursor().head(), 0);
        assert!(!consumer.avail_cursor().wrap());
    }

    #[test]
    fn p0016_queue_size_one_next_chain_is_rejected() {
        let ring = make_ring(1);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::NEXT);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0017_stale_driver_event_wrap_suppresses_notification() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, 0, EventFlags::DESC.bits());
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, _) = consumer.poll_available().unwrap();

        assert!(!consumer.submit_used_with_notify(id, 0).unwrap());
    }

    #[test]
    fn p0018_pre_used_descriptor_is_ignored() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let mut desc = raw_desc(0, 16, DescFlags::empty());
        desc.mark_used(true);
        ring.write_desc(0, desc);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0019_driver_event_wrap_mismatch_suppresses_notification() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, 1, EventFlags::DESC.bits());
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, _) = consumer.poll_available().unwrap();

        assert!(!consumer.submit_used_with_notify(id, 0).unwrap());
    }

    #[test]
    fn p0020_ring_full_reports_backpressure() {
        let ring = make_ring(4);
        let mut producer = make_producer(&ring);

        for idx in 0..4 {
            producer.submit_one(0x1000 + idx * 0x10, 16, false).unwrap();
        }

        assert!(matches!(
            producer.submit_one(0x2000, 16, false),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0022_indirect_reused_id_is_rejected_before_id_reuse() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::INDIRECT);
        write_avail(&ring, 1, 0, DescFlags::empty());

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0023_desc_event_mode_suppresses_before_later_slot() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, 0x8001, EventFlags::DESC.bits());
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, _) = consumer.poll_available().unwrap();

        assert!(!consumer.submit_used_with_notify(id, 0).unwrap());
    }

    #[test]
    fn p0024_single_slot_indirect_is_rejected() {
        let ring = make_ring(1);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::INDIRECT);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0025_used_completions_can_arrive_out_of_order() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let id0 = producer.submit_one(0x1000, 16, true).unwrap();
        let id1 = producer.submit_one(0x2000, 16, true).unwrap();
        let id2 = producer.submit_one(0x3000, 16, true).unwrap();

        let (dev0, _) = consumer.poll_available().unwrap();
        let (dev1, _) = consumer.poll_available().unwrap();
        let (dev2, _) = consumer.poll_available().unwrap();

        consumer.submit_used(dev2, 32).unwrap();
        consumer.submit_used(dev0, 16).unwrap();
        consumer.submit_used(dev1, 24).unwrap();

        assert_eq!(producer.poll_used().unwrap().id, id2);
        assert_eq!(producer.poll_used().unwrap().id, id0);
        assert_eq!(producer.poll_used().unwrap().id, id1);
    }

    #[test]
    fn p0026_all_descriptor_flags_set_is_not_available() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        ring.write_desc(0, raw_desc(0, 16, DescFlags::from_bits_truncate(u16::MAX)));

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0028_multi_wrap_round_trip_remains_ordered() {
        let ring = make_ring(2);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        for idx in 0..8u64 {
            let id = producer.submit_one(0x1000 + idx * 0x10, 16, false).unwrap();
            let (dev_id, _) = consumer.poll_available().unwrap();
            assert_eq!(dev_id, id);
            consumer.submit_used(dev_id, idx as u32).unwrap();
            assert_eq!(producer.poll_used().unwrap().id, id);
        }
    }

    #[test]
    fn p0029_desc_event_idx_zero_notifies() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, 0x8000, EventFlags::DESC.bits());
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, _) = consumer.poll_available().unwrap();

        assert!(consumer.submit_used_with_notify(id, 0).unwrap());
    }

    #[test]
    fn p0031_reserved_driver_event_flags_suppress_notification() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, 0, 3);
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, _) = consumer.poll_available().unwrap();

        assert!(!consumer.submit_used_with_notify(id, 0).unwrap());
    }

    #[test]
    fn p0032_unaligned_indirect_descriptor_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let mut desc = Descriptor::new(0x1003, 16, 0, DescFlags::INDIRECT);
        desc.mark_avail(true);
        ring.write_desc(0, desc);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0033_scatter_gather_chain_is_preserved() {
        let ring = make_ring(16);
        let mut consumer = make_consumer(&ring);

        for idx in 0..8u16 {
            let flags = if idx == 7 {
                DescFlags::WRITE
            } else {
                DescFlags::NEXT
            };
            write_avail(&ring, idx, 0, flags);
        }

        let (_, chain) = consumer.poll_available().unwrap();
        assert_eq!(chain.len(), 8);
        assert_eq!(chain.readables().len(), 7);
        assert_eq!(chain.writables().len(), 1);
    }

    #[test]
    fn p0034_page_boundary_indirect_descriptor_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let mut desc = Descriptor::new(0x1ff0, 16, 0, DescFlags::INDIRECT);
        desc.mark_avail(true);
        ring.write_desc(0, desc);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0035_both_avail_and_used_clear_is_ignored() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        ring.write_desc(0, raw_desc(0, 16, DescFlags::empty()));

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0037_event_flags_can_cycle_between_batches() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        for (idx, (flags, expect_notify)) in [
            (EventFlags::ENABLE.bits(), true),
            (EventFlags::DISABLE.bits(), false),
            (EventFlags::ENABLE.bits(), true),
        ]
        .into_iter()
        .enumerate()
        {
            write_driver_event(&ring, 0, flags);
            write_avail(&ring, idx as u16, idx as u16, DescFlags::empty());
            let (id, _) = consumer.poll_available().unwrap();
            assert_eq!(
                consumer.submit_used_with_notify(id, 0).unwrap(),
                expect_notify
            );
        }
    }

    #[test]
    fn p0038_many_wraps_on_minimum_queue_size() {
        let ring = make_ring(1);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        for idx in 0..8u64 {
            let id = producer.submit_one(0x1000 + idx * 0x10, 16, false).unwrap();
            let (dev_id, _) = consumer.poll_available().unwrap();
            assert_eq!(dev_id, id);
            consumer.submit_used(dev_id, 0).unwrap();
            assert_eq!(producer.poll_used().unwrap().id, id);
        }
    }

    #[test]
    fn p0039_reserved_device_event_flags_suppress_avail_notification() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        write_device_event(&ring, 0, 3);

        assert!(
            !producer
                .submit_one_with_notify(0x1000, 16, false)
                .unwrap()
                .notify
        );
    }

    #[test]
    fn p0040_indirect_with_next_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::INDIRECT | DescFlags::NEXT);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0041_driver_event_offset_at_queue_size_suppresses_notification() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_driver_event(&ring, ring.len() as u16, EventFlags::DESC.bits());
        write_avail(&ring, 0, 0, DescFlags::empty());

        let (id, _) = consumer.poll_available().unwrap();

        assert!(!consumer.submit_used_with_notify(id, 0).unwrap());
    }

    #[test]
    fn p0042_empty_ring_has_no_available_work() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0043_id_equal_to_queue_size_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, ring.len() as u16, DescFlags::empty());

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::InvalidState)
        ));
    }

    #[test]
    fn p0044_avail_and_used_set_is_not_available() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        ring.write_desc(0, raw_desc(0, 16, DescFlags::AVAIL | DescFlags::USED));

        assert!(!consumer.peek_available().unwrap());
        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0045_zero_length_descriptor_is_consumable() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail_len(&ring, 0, 0, 0, DescFlags::WRITE);

        let (id, chain) = consumer.poll_available().unwrap();
        assert_eq!(id, 0);
        assert_eq!(chain.writables()[0].len, 0);
    }

    #[test]
    fn p0046_max_length_descriptor_is_consumable_at_ring_layer() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail_len(&ring, 0, 0, u32::MAX, DescFlags::WRITE);

        let (id, chain) = consumer.poll_available().unwrap();
        assert_eq!(id, 0);
        assert_eq!(chain.writables()[0].len, u32::MAX);
    }

    #[test]
    fn p0047_zero_descriptor_address_is_preserved_at_ring_layer() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        let mut desc = Descriptor::new(0, 16, 0, DescFlags::WRITE);
        desc.mark_avail(true);
        ring.write_desc(0, desc);

        let (id, chain) = consumer.poll_available().unwrap();
        assert_eq!(id, 0);
        assert_eq!(chain.writables()[0].addr, 0);
    }

    #[test]
    fn p0049_indirect_entry_with_phase_bits_is_rejected() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        ring.write_desc(
            0,
            raw_desc(
                0,
                16,
                DescFlags::INDIRECT | DescFlags::AVAIL | DescFlags::USED,
            ),
        );

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0050_nested_indirect_is_rejected_at_head() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::INDIRECT);

        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::BadChain)
        ));
    }

    #[test]
    fn p0051_write_flag_controls_readable_writable_split() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::NEXT);
        write_avail(&ring, 1, 0, DescFlags::WRITE);

        let (_, chain) = consumer.poll_available().unwrap();
        assert_eq!(chain.readables().len(), 1);
        assert_eq!(chain.writables().len(), 1);
    }

    #[test]
    fn p0052_batch_completion_signals_each_chain_head() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let id0 = producer.submit_one(0x1000, 16, true).unwrap();
        let id1 = producer.submit_one(0x2000, 16, true).unwrap();

        let (dev0, _) = consumer.poll_available().unwrap();
        let (dev1, _) = consumer.poll_available().unwrap();
        consumer.submit_used(dev0, 16).unwrap();
        consumer.submit_used(dev1, 32).unwrap();

        assert_eq!(producer.poll_used().unwrap().id, id0);
        assert_eq!(producer.poll_used().unwrap().id, id1);
    }

    #[test]
    fn p0053_descriptor_without_avail_bit_is_ignored() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        ring.write_desc(0, raw_desc(0, 16, DescFlags::WRITE));

        assert!(!consumer.peek_available().unwrap());
        assert!(matches!(
            consumer.poll_available(),
            Err(RingError::WouldBlock)
        ));
    }

    #[test]
    fn p0054_event_suppression_struct_layout_matches_packed_ring() {
        assert_eq!(EventSuppression::SIZE, 4);
        assert_eq!(EventSuppression::WRAP_OFFSET, 0);
        assert_eq!(EventSuppression::FLAGS_OFFSET, 2);
    }

    #[test]
    fn p0055_descriptor_after_chain_tail_is_ignored() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        write_avail(&ring, 0, 0, DescFlags::empty());
        write_avail(&ring, 1, 1, DescFlags::WRITE);

        let (id, chain) = consumer.poll_available().unwrap();
        assert_eq!(id, 0);
        assert_eq!(chain.len(), 1);
    }

    #[test]
    fn p0056_next_chain_order_is_preserved() {
        let ring = make_ring(8);
        let mut consumer = make_consumer(&ring);

        for idx in 0..3u16 {
            let mut desc = Descriptor::new(
                0x1000 + u64::from(idx) * 0x100,
                16,
                0,
                if idx == 2 {
                    DescFlags::WRITE
                } else {
                    DescFlags::NEXT
                },
            );
            desc.mark_avail(true);
            ring.write_desc(idx, desc);
        }

        let (_, chain) = consumer.poll_available().unwrap();
        assert_eq!(chain.elems()[0].addr, 0x1000);
        assert_eq!(chain.elems()[1].addr, 0x1100);
        assert_eq!(chain.elems()[2].addr, 0x1200);
    }

    #[test]
    fn p0057_device_event_enable_requests_avail_notification() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);

        write_device_event(&ring, 0, EventFlags::ENABLE.bits());

        assert!(
            producer
                .submit_one_with_notify(0x1000, 16, false)
                .unwrap()
                .notify
        );
    }

    #[test]
    fn p0058_used_descriptor_carries_id_and_len() {
        let ring = make_ring(8);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        let id = producer.submit_one(0x1000, 16, true).unwrap();
        let (dev_id, _) = consumer.poll_available().unwrap();
        consumer.submit_used(dev_id, 42).unwrap();

        let used = producer.poll_used().unwrap();
        assert_eq!(used.id, id);
        assert_eq!(used.len, 42);
    }
}

#[cfg(test)]
mod fuzz {
    use quickcheck::{Arbitrary, Gen, QuickCheck};

    use super::tests::{OwnedRing, make_consumer, make_producer};
    use super::*;

    const MAX_RING: usize = 64;
    const MAX_OPS: usize = 128;
    const MAX_CHAIN_LEN: usize = 8;

    #[allow(clippy::large_enum_variant)]
    #[derive(Clone, Debug)]
    enum Op {
        /// submit one chain
        Submit(BufferChain),
        /// poll up to N chains
        PollAvail(u8),
        /// driver reclaims up to N completions
        PollUsed(u8),
        /// complete one previously polled chain
        CompleteOne,
    }

    impl Arbitrary for Op {
        fn arbitrary(g: &mut Gen) -> Self {
            let choice = u8::arbitrary(g) % 4;
            match choice {
                0 => Op::Submit(BufferChain::arbitrary(g)),
                1 => Op::PollAvail(u8::arbitrary(g) % 8 + 1),
                2 => Op::PollUsed(u8::arbitrary(g) % 8 + 1),
                3 => Op::CompleteOne,
                _ => unreachable!(),
            }
        }
    }

    #[derive(Clone, Debug)]
    struct Scenario {
        table_size: usize,
        ops: Vec<Op>,
    }

    impl Arbitrary for Scenario {
        fn arbitrary(g: &mut Gen) -> Self {
            let table_size = (usize::arbitrary(g) % MAX_RING + 1).next_power_of_two();
            let num_ops = usize::arbitrary(g) % MAX_OPS + 1;

            let ops = (0..num_ops).map(|_| Op::arbitrary(g)).collect();
            Scenario { table_size, ops }
        }
    }

    impl Arbitrary for BufferElement {
        fn arbitrary(g: &mut Gen) -> Self {
            let addr = u64::arbitrary(g);
            let len = u32::arbitrary(g);
            let writable = bool::arbitrary(g);

            BufferElement {
                addr,
                len,
                writable,
            }
        }
    }

    impl Arbitrary for BufferChain {
        fn arbitrary(g: &mut Gen) -> Self {
            let chain_len = usize::arbitrary(g) % MAX_CHAIN_LEN + 1;

            let mut elems = vec![BufferElement::zeroed(); chain_len];
            let mut readables = 0;
            let mut writables = 0;

            for _ in 0..chain_len {
                let elem = BufferElement::arbitrary(g);
                if elem.writable {
                    elems[chain_len - 1 - writables] = elem;
                    writables += 1;
                } else {
                    elems[readables] = elem;
                    readables += 1;
                }
            }

            BufferChain {
                elems: elems.into(),
                split: readables,
            }
        }
    }

    fn run_scenario(s: Scenario) -> bool {
        let ring = OwnedRing::new(s.table_size);
        let mut producer = make_producer(&ring);
        let mut consumer = make_consumer(&ring);

        // Order logs
        let mut dev_order: Vec<u16> = Vec::new();
        let mut drv_order: Vec<u16> = Vec::new();

        // Device-tracked polled-but-not-completed IDs
        let mut dev_ready: Vec<(u16, u32)> = Vec::new();

        for op in &s.ops {
            match op {
                Op::Submit(chain) => {
                    // Submit only if space; otherwise skip
                    let _ = producer.submit_available(chain);
                }
                Op::PollAvail(n) => {
                    for _ in 0..*n {
                        if let Ok((id, chain)) = consumer.poll_available() {
                            dev_ready.push((id, chain.len() as u32));
                        } else {
                            break;
                        }
                    }
                }
                Op::PollUsed(n) => {
                    for _ in 0..*n {
                        match producer.poll_used() {
                            Ok(u) => {
                                drv_order.push(u.id);
                                if producer.id_num[u.id as usize] != 0 {
                                    return false;
                                }
                                if !producer.id_free.contains(&u.id) {
                                    return false;
                                }
                            }
                            Err(RingError::WouldBlock) => break,
                            Err(_) => return false,
                        }
                    }
                }
                Op::CompleteOne => {
                    if let Some((id, len)) = dev_ready.pop() {
                        if consumer.submit_used(id, len).is_err() {
                            return false;
                        }

                        dev_order.push(id);
                    }
                }
            }

            // assert invariants after each op
            let outstanding: u16 = producer.id_num.iter().copied().sum();
            if outstanding as usize + producer.num_free != ring.len() {
                return false;
            }

            for id in producer.id_free.iter() {
                if producer.id_num[*id as usize] != 0 {
                    return false;
                }
            }
        }

        // Drain remaining completions and reclaims
        while let Some((id, len)) = dev_ready.pop() {
            if consumer.submit_used(id, len).is_err() {
                return false;
            }
        }

        loop {
            match producer.poll_used() {
                Ok(u) => drv_order.push(u.id),
                Err(RingError::WouldBlock) => break,
                Err(_) => return false,
            }
        }

        true
    }

    #[test]
    fn prop_interleaved_with_order_verification() {
        #[cfg(miri)]
        let tests = 1;
        #[cfg(not(miri))]
        let tests = 100;

        QuickCheck::new()
            .tests(tests)
            .quickcheck(run_scenario as fn(Scenario) -> bool);
    }
}