windows-ioring-sys 0.7.2

// Copyright (c) 2026 Mike Grier
//! `Batch`: a scoped, exclusive submission window (M3.1-M3.3, M3.5, M5).

use std::ffi::c_void;
use std::io;
use std::mem::ManuallyDrop;
use std::os::windows::io::AsRawHandle;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use windows_sys::Win32::Foundation::HANDLE;
use windows_sys::Win32::Storage::FileSystem::{
    FILE_FLUSH_DATA, FILE_FLUSH_DEFAULT, FILE_FLUSH_MIN_METADATA, FILE_FLUSH_MODE,
    FILE_FLUSH_NO_SYNC, FILE_WRITE_FLAGS, FILE_WRITE_FLAGS_NONE, FILE_WRITE_FLAGS_WRITE_THROUGH,
    IORING_BUFFER_INFO, IORING_BUFFER_REF, IORING_BUFFER_REF_0, IORING_HANDLE_REF,
    IORING_HANDLE_REF_0, IORING_REF_RAW, IORING_REF_REGISTERED, IORING_REGISTERED_BUFFER,
    IORING_SQE_FLAGS, IOSQE_FLAGS_DRAIN_PRECEDING_OPS, IOSQE_FLAGS_NONE,
};

use crate::error::{PushRefused, check};
use crate::ring::{AllowsRaw, Completion, Entry, Held, IoRing, Op, Open, RingId};
use crate::token::OperationId;
use io_buffers::{IoBuf, IoBufMut, RegisteredSpan};
use win_shared_os_owned_handle::SharedHandle;

/// Per-push options shared across every op builder (M3.2).
#[derive(Clone, Copy, Debug, Default)]
#[must_use]
pub struct PushOptions {
    drain_preceding: bool,
}

impl PushOptions {
    /// The default options: no barrier.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set `IOSQE_FLAGS_DRAIN_PRECEDING_OPS`: this op does not complete until
    /// every op already queued on this batch's ring has completed. A
    /// barrier, not a cheap tag -- but a **one-sided** one, and the direction
    /// it does not cover is the surprising half.
    ///
    /// Stated in terms of completion rather than of when the op *starts*,
    /// throughout: user mode cannot see execution begin, only completions
    /// posted, so completion order is both what was measured and the only
    /// thing a caller can act on ([D-47](../DESIGN-NOTES.md#d-47-detail)).
    ///
    /// # Ring-wide in reach, one-sided in effect
    ///
    /// Measured, not inferred (D-24 and D-47 in `DESIGN-NOTES.md`). The wait
    /// reaches every operation outstanding on the *ring*, not only the ones
    /// queued in this batch, and results were identical whether the sequence
    /// went in one [`Batch::submit`] or three.
    ///
    /// **It does not hold back operations pushed after it.** Those can complete
    /// while this one is still outstanding -- observed across about 4,500
    /// trials, and in the worst case every one of 32 subsequent writes
    /// completed first. An earlier version of this documentation said they were
    /// held; that was measured over too few runs to see a fault this rare, and
    /// it was wrong.
    ///
    /// The consequence to plan for: **this delays the flagged op, not the
    /// ring.** Later work is not blocked, so it cannot be used to fence
    /// anything that follows. A consumer that needs later work to begin only
    /// after this op is durable must sequence that itself, by waiting for its
    /// completion before submitting.
    ///
    /// # The barrier stops at the ring's edge
    ///
    /// It orders SQEs against SQEs, and nothing else. Any non-ring I/O -- an
    /// overlapped `DeviceIoControl`, anything issued through
    /// `windows-overlapped-io-sys`, a plain blocking write -- is outside it in
    /// *both* directions: this flag can neither make a ring op wait for a
    /// non-ring op nor make a non-ring op wait for ring ops.
    ///
    /// A consumer mixing both paths is the normal case rather than an exotic
    /// one, and it must enforce that ordering in its own code. What this
    /// crate offers toward it is [`IoRing::completion_event`], which makes
    /// waiting on the ring *alongside* other handles expressible without
    /// giving up the ring or blocking a thread in a drain.
    ///
    /// # Ordering, never durability by itself
    ///
    /// This flag orders operations; it does not flush anything. It is,
    /// however, what makes a flush cover the writes before it: a flush pushed
    /// after a batch of writes *without* this flag routinely completes while
    /// many of those writes are still outstanding (D-23 in `DESIGN-NOTES.md`,
    /// measured at 17 and 23 of 32 writes finishing after the flush did). So
    /// the obvious spelling -- push the writes, then push a flush -- silently
    /// does not make those writes durable.
    ///
    /// A flush does not take its barrier decision from here, for exactly that
    /// reason: [`Batch::flush_owned`] requires a [`FlushCoverage`] instead, so the
    /// choice cannot be inherited from a default (M12.1, D-25).
    pub fn drain_preceding(mut self, drain: bool) -> Self {
        self.drain_preceding = drain;
        self
    }

    fn sqe_flags(self) -> IORING_SQE_FLAGS {
        if self.drain_preceding {
            IOSQE_FLAGS_DRAIN_PRECEDING_OPS
        } else {
            IOSQE_FLAGS_NONE
        }
    }
}

/// Whether a flush covers the operations queued before it (M12.1, D-23 in
/// `DESIGN-NOTES.md`).
///
/// This is a **required argument** on [`Batch::flush_owned`] and
/// [`Batch::flush_raw`] rather than a field of [`PushOptions`], because there
/// is no defensible default. An unflagged flush does *not* cover preceding
/// writes: it is an ordinary operation competing with them, and it frequently
/// wins. A default would make the obvious spelling -- push the writes, then
/// push a flush -- a silent data-loss bug rather than a missing feature, so
/// the decision is taken at the call site instead (D-25).
///
/// It is an enum rather than a `bool` for the same reason: `flush(&file,
/// true)` does not say what the `true` decides, and this is not a parameter
/// anyone should have to look up.
///
/// # The barrier's scope is the ring; the flush's is the file
///
/// One call sets both, which makes them easy to merge, and merging them is a
/// mistake about durability rather than about style:
///
/// - **This flag is ring-wide.** [`Self::CoversPrecedingOperations`] means the
///   flush does not execute until every operation outstanding *on the ring*
///   when it was reached has **completed** -- whatever file each one targets,
///   and whoever queued it (D-47, measured over roughly 4,500 trials).
/// - **The flush names one file.** [`Batch::flush_owned`] takes a [`FileTarget`], so
///   what a syncing [`FlushMode`] pushes to stable media is that file's data
///   and the device cache behind it.
///
/// **Completion is not durability.** A write completing means the kernel took
/// the bytes, not that they reached non-volatile media. So the barrier bounds
/// what a flush **waits for**, and the flush itself bounds what is **made
/// durable**, and those are different sets whenever a ring carries operations
/// against more than one file.
///
/// The practical consequence for a caller is cost rather than correctness:
/// operations this flush will never make durable can still make it wait. A
/// caller who cares about that bounds it by controlling what shares the ring;
/// this crate does not decide that (D-8), and the flush's own target is what
/// secures the durability of the file named.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FlushCoverage {
    /// Wait for every operation already outstanding on the ring, then flush.
    ///
    /// This is what makes preceding writes durable when the flush's
    /// completion is observed, and it is what a caller closing an epoch
    /// wants.
    ///
    /// It sets `IOSQE_FLAGS_DRAIN_PRECEDING_OPS`, whose reach is **ring-wide**
    /// rather than per-file: it waits on every operation outstanding on the
    /// ring when it is reached, not only the ones in this batch. So the flush
    /// takes as long as the slowest of them, however unrelated. That cost is
    /// real, and it is the only reason the other variant exists.
    ///
    /// # What this does not promise
    ///
    /// **The flag is one-sided. Operations you push *after* this flush are not
    /// held back, and can complete before it.** Measured over about 4,500
    /// trials at a rate between 0.03% and 0.8%, and in the worst case every one
    /// of 32 subsequent writes completed first (D-47).
    ///
    /// So this flush's completion tells you about the past and not the future:
    /// everything outstanding when the flush was reached is durable, and
    /// nothing is implied about work submitted afterwards. If later work must
    /// not begin until the flush is durable, submit it after you have popped
    /// the flush's completion. The ring will not sequence it for you.
    ///
    /// An earlier version of this documentation said subsequent operations were
    /// held until the flush completed. That was wrong, and it is corrected here
    /// rather than quietly dropped because a caller may have relied on it.
    CoversPrecedingOperations,
    /// Queue the flush with no barrier: it may complete while writes pushed
    /// before it are still in flight.
    ///
    /// **Almost never what a caller wants.** Its completion proves nothing
    /// about any preceding write, so using it to close an epoch loses data
    /// that the caller believes is committed -- invisibly, until power is
    /// lost.
    ///
    /// Two uses are legitimate. One is *host sequencing*: the caller has
    /// already observed the completions of every write in the epoch before
    /// pushing this, so the ordering is established outside the ring and the
    /// barrier would only make this flush wait on operations it has already
    /// accounted for. The other is a flush that is not being used for
    /// durability at all.
    Unordered,
}

impl FlushCoverage {
    fn sqe_flags(self) -> IORING_SQE_FLAGS {
        match self {
            Self::CoversPrecedingOperations => IOSQE_FLAGS_DRAIN_PRECEDING_OPS,
            Self::Unordered => IOSQE_FLAGS_NONE,
        }
    }
}

/// Where a write is allowed to stop: in the system cache, or past it (M12.3,
/// D-25 in `DESIGN-NOTES.md`).
///
/// This is `BuildIoRingWriteFile`'s `FILE_WRITE_FLAGS` parameter, which the
/// crate previously hardcoded to `FILE_WRITE_FLAGS_NONE` -- so a consumer
/// reading this API saw ordering but no way to express anything about caching
/// at all, and could reasonably conclude the ring did not expose it. It does;
/// the crate was narrowing the platform to what its own examples needed.
///
/// # What write-through is, and what it is not
///
/// Worth stating precisely, because conflating these produced a wrong
/// recommendation in the exchange that prompted this work.
///
/// It **is** a first-level cache directive: it tells the OS not to leave the
/// data sitting in the system cache. Its practical value is latency shaping --
/// data already at the device makes a later flush shorter, which can matter
/// when the flush is on a commit path.
///
/// It is **not** a durability guarantee, and it is **not** FUA. Whether it
/// becomes a Force Unit Access bit on the underlying command depends on the
/// driver, the volume, and whether the device's write cache is enabled -- none
/// of which this API can see or promise. A write that completes with
/// [`WriteCaching::WriteThrough`] may still be sitting in a volatile device
/// cache, and will be lost on power failure like any other.
///
/// **The flush operation is the only durability primitive the ring has**, and
/// only with [`FlushCoverage::CoversPrecedingOperations`]. Reaching for
/// write-through instead of a covering flush is the mistake this doc exists to
/// prevent; reaching for it *as well* is a legitimate latency optimization.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum WriteCaching {
    /// `FILE_WRITE_FLAGS_NONE`: the write may be satisfied into the system
    /// cache, and reaching the device is a later flush's job.
    ///
    /// The default, and the right choice for the streaming writes of an epoch
    /// that a covering flush will close.
    #[default]
    Cached,
    /// `FILE_WRITE_FLAGS_WRITE_THROUGH`: ask the OS not to leave the data in
    /// the system cache.
    ///
    /// A latency-shaping knob, not a durability marker -- see this type's
    /// docs. It does not remove the need for a flush.
    WriteThrough,
}

impl WriteCaching {
    fn raw(self) -> FILE_WRITE_FLAGS {
        match self {
            Self::Cached => FILE_WRITE_FLAGS_NONE,
            Self::WriteThrough => FILE_WRITE_FLAGS_WRITE_THROUGH,
        }
    }
}

/// How much a flush is asked to push, and whether it syncs the device at all
/// (M12.4, D-25 in `DESIGN-NOTES.md`).
///
/// This is `BuildIoRingFlushFile`'s `FILE_FLUSH_MODE` parameter, which the
/// crate previously hardcoded to `FILE_FLUSH_DEFAULT`. Exposing it is the
/// other half of D-25: the kernel offers a durability parameter here and a
/// wrapper that hides it narrows the platform to what its own examples needed.
///
/// [`FlushMode::Default`] is what a caller closing an epoch wants, and is
/// what the crate did before this existed.
///
/// # `NoSync` is not a durability operation
///
/// Three of these four modes issue the device sync; [`FlushMode::NoSync`] does
/// not. That is worth stating rather than inferring, and the inference runs
/// the other way too: **the existence of a distinct "no sync" mode is the
/// evidence that the other three do sync.** Nothing in the Win32
/// documentation says so directly.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum FlushMode {
    /// `FILE_FLUSH_DEFAULT`: flush the file's data and metadata, and sync the
    /// device.
    ///
    /// The default, and the mode a durability barrier wants. Pair it with
    /// [`FlushCoverage::CoversPrecedingOperations`] or it covers nothing.
    #[default]
    Default,
    /// `FILE_FLUSH_DATA`: flush the file's data, and sync the device.
    ///
    /// Metadata that is not required to read the data back may be left
    /// unflushed. The analogue of `fdatasync`, and the usual choice for a log
    /// whose records are self-describing.
    Data,
    /// `FILE_FLUSH_MIN_METADATA`: flush the data plus the minimum metadata
    /// needed to retrieve it, and sync the device.
    MinMetadata,
    /// `FILE_FLUSH_NO_SYNC`: flush to the device **without** issuing the sync.
    ///
    /// **This is the one mode that makes nothing durable.** It pushes data out
    /// of the system cache and stops there, so anything still held in a
    /// volatile device cache is lost on power failure exactly as if no flush
    /// had been issued at all. A completion from this mode is not a commit
    /// point and must never be reported to a caller as one.
    ///
    /// It is useful for shaping when cache pressure is paid -- moving data
    /// deviceward early so a later real flush is short -- which is the same
    /// role [`WriteCaching::WriteThrough`] plays on the write side, and it
    /// carries the same warning: a latency knob, not a durability primitive.
    NoSync,
}

impl FlushMode {
    fn raw(self) -> FILE_FLUSH_MODE {
        match self {
            Self::Default => FILE_FLUSH_DEFAULT,
            Self::Data => FILE_FLUSH_DATA,
            Self::MinMetadata => FILE_FLUSH_MIN_METADATA,
            Self::NoSync => FILE_FLUSH_NO_SYNC,
        }
    }
}

/// Build the raw handle reference for `file`, refusing a [`RegisteredFile`]
/// that did not come from `ring_id`'s own ring (PR #20 review response): the
/// index alone names a slot in *some* ring's file table, and a slot from a
/// different ring can legitimately hold an entirely different file, so
/// crossing rings must be an explicit, reported error rather than silently
/// addressing the wrong handle.
fn handle_ref(file: FileRef, ring_id: RingId) -> io::Result<IORING_HANDLE_REF> {
    match file {
        FileRef::Raw(handle) => Ok(IORING_HANDLE_REF {
            Kind: IORING_REF_RAW,
            Handle: IORING_HANDLE_REF_0 { Handle: handle },
        }),
        FileRef::Registered(file) => {
            if file.ring_id != ring_id {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "this RegisteredFile was registered on a different IoRing",
                ));
            }
            Ok(IORING_HANDLE_REF {
                Kind: IORING_REF_REGISTERED,
                Handle: IORING_HANDLE_REF_0 {
                    Index: file.index(),
                },
            })
        }
    }
}

fn raw_buffer_ref(address: *mut c_void) -> IORING_BUFFER_REF {
    IORING_BUFFER_REF {
        Kind: IORING_REF_RAW,
        Buffer: IORING_BUFFER_REF_0 { Address: address },
    }
}

fn registered_buffer_ref(index: u32, offset: u32) -> IORING_BUFFER_REF {
    IORING_BUFFER_REF {
        Kind: IORING_REF_REGISTERED,
        Buffer: IORING_BUFFER_REF_0 {
            IndexAndOffset: IORING_REGISTERED_BUFFER {
                BufferIndex: index,
                Offset: offset,
            },
        },
    }
}

fn checked_len(len: usize) -> io::Result<u32> {
    u32::try_from(len).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("an IoRing buffer is limited to u32::MAX bytes; {len} does not fit"),
        )
    })
}

/// A file, addressed either by a raw `HANDLE` or by the index a prior
/// [`Batch::register_files`] assigned it (M5.1).
///
/// This enum is the `_raw` pushes' addressing parameter. A safe push takes a
/// [`FileTarget`] instead -- a [`SharedHandle`] or a [`RegisteredFile`] -- so
/// reaching a registered file has not required `unsafe` since M10.4 (D-29,
/// D-33). Constructing a `FileRef` directly is only necessary for the `_raw`
/// pushes, whose reason to exist is a bare `HANDLE` this crate cannot keep
/// alive for you.
///
/// A distinct type per addressing mode -- rather than one method accepting
/// either a `HANDLE` or a bare `u32` -- is what makes "read against a
/// registered file" and "read against an unregistered one" impossible to
/// confuse by accident: a [`RegisteredFile`] cannot be mistaken for a raw
/// `HANDLE`, or vice versa, because they are different types.
#[derive(Clone, Copy, Debug)]
pub enum FileRef {
    /// `IORING_REF_RAW`: the caller's own open handle.
    Raw(HANDLE),
    /// `IORING_REF_REGISTERED`: an index from [`Batch::register_files`].
    Registered(RegisteredFile),
}

impl From<HANDLE> for FileRef {
    fn from(handle: HANDLE) -> Self {
        FileRef::Raw(handle)
    }
}

impl From<RegisteredFile> for FileRef {
    fn from(file: RegisteredFile) -> Self {
        FileRef::Registered(file)
    }
}

/// One index a [`Batch::register_files`] registration assigned (M5.1).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RegisteredFile {
    index: u32,
    ring_id: RingId,
}

impl RegisteredFile {
    /// The raw registered-file index.
    #[must_use]
    pub fn index(self) -> u32 {
        self.index
    }
}

mod sealed {
    pub trait Sealed {}
    impl Sealed for super::SharedHandle {}
    impl Sealed for super::RegisteredFile {}
}

/// A file a *safe* push can address: either a [`SharedHandle`] or a
/// [`RegisteredFile`] (M10.4; D-29, D-33 and D-82 in `DESIGN-NOTES.md`).
///
/// This exists because the two carry different lifetime obligations, and
/// exactly one of them has none. A `SharedHandle` is a raw `HANDLE` underneath,
/// so something must keep it open until the kernel is finished; a
/// `RegisteredFile` is an index into a table the ring itself owns, minted by
/// this crate and checked against the minting ring, so there is nothing for a
/// caller to keep alive per push. (The handles behind that table are kept
/// open for the ring's life either by [`Batch::register_files`]' caller, under
/// its `unsafe` contract, or by the ring itself after
/// [`Batch::register_shared_files`].) Both are therefore safe to push, and neither needs
/// `unsafe` -- which is why [`Batch::read_owned`] and its siblings are generic over
/// this trait rather than hardcoding `SharedHandle`.
///
/// [`Guard`](FileTarget::Guard) is what the ring holds until the
/// operation's completion is observed. For `SharedHandle` that is a clone,
/// which shares the handle and so makes it outlive the operation; for
/// `RegisteredFile` there is nothing to keep alive, so it is the (`Copy`)
/// index itself, handed back for symmetry rather than out of necessity.
///
/// **Sealed**: this crate implements it for exactly those two types and no
/// others. An outside implementation could return an arbitrary raw `HANDLE`
/// from [`as_file_ref`](FileTarget::as_file_ref) with no guard keeping it
/// alive, which is precisely the unsoundness the `_raw` pushes make a caller
/// take responsibility for with `unsafe`.
pub trait FileTarget: sealed::Sealed {
    /// What the ring must hold for the operation until its completion is
    /// observed.
    /// What keeps the file valid while the operation runs.
    ///
    /// `Into<FileGuard>` is expressible only because this trait is **sealed**
    /// (`D-73`): both implementors are this crate's, so the ring can hold a
    /// guard in a concrete slot instead of a third generic parameter every
    /// consumer would have to name.
    type Guard: Send + 'static + Into<crate::ring::FileGuard>;

    /// How this target addresses its file.
    fn as_file_ref(&self) -> FileRef;

    /// Produce the value the ring will hold for the operation's
    /// duration.
    fn guard(&self) -> Self::Guard;
}

/// A [`SharedHandle`] is how a caller gives this crate a file it does not have
/// to prove outlives every operation pushed against it (M8; D-82). One handle is
/// legitimately the target of many concurrent pushes, so what must survive until
/// every one of them completes is a *reference*, not sole ownership. Every safe
/// push (e.g. [`Batch::read_owned`], as opposed to its `_raw` sibling) clones it
/// into the same inventory entry that tracks the operation's own payload, so the
/// handle survives until the ring returns that entry at its pop (D-4), whatever
/// the caller does with its own clone.
impl FileTarget for SharedHandle {
    type Guard = SharedHandle;

    fn as_file_ref(&self) -> FileRef {
        FileRef::Raw(self.as_raw_handle())
    }

    fn guard(&self) -> Self::Guard {
        self.clone()
    }
}

impl FileTarget for RegisteredFile {
    type Guard = RegisteredFile;

    fn as_file_ref(&self) -> FileRef {
        FileRef::Registered(*self)
    }

    fn guard(&self) -> Self::Guard {
        *self
    }
}

/// The confirmed result of a [`Batch::register_files`] push: `count`
/// contiguous indices starting at `base_index`.
///
/// `base_index` is currently always zero, because a ring accepts at most one
/// registration that assigns an index -- do not read its presence as evidence
/// that several registrations compose. It is kept rather than folded away
/// because it is the correct shape if that rule is ever relaxed (M10.3, D-31).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RegisteredFiles {
    base_index: u32,
    count: u32,
    ring_id: RingId,
}

impl RegisteredFiles {
    /// How many handles this registration covers.
    #[must_use]
    pub fn len(&self) -> u32 {
        self.count
    }

    /// Whether this registration covers no handles.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.count == 0
    }

    /// The [`RegisteredFile`] for the `i`-th handle passed to
    /// [`Batch::register_files`], or `None` if `i` is out of range.
    #[must_use]
    pub fn get(&self, i: u32) -> Option<RegisteredFile> {
        (i < self.count).then(|| RegisteredFile {
            index: self.base_index + i,
            ring_id: self.ring_id,
        })
    }
}

/// A [`Batch::register_files`] push not yet matched to its completion.
#[derive(Debug)]
pub struct PendingFileRegistration {
    user_data: usize,
    base_index: u32,
    count: u32,
    ring_id: RingId,
}

impl PendingFileRegistration {
    /// This push's `UserData` identity, to match against a popped
    /// [`Completion`].
    #[must_use]
    pub fn user_data(&self) -> usize {
        self.user_data
    }

    /// Turn this pending registration into usable [`RegisteredFiles`] once
    /// `completion` names it, or hand it back unchanged if `completion`
    /// names a different operation.
    ///
    /// # Errors
    ///
    /// The inner `Result` is `Err` if the registration itself failed --
    /// nothing is lost, since no owned resource was ever handed over for
    /// this op (M5.1, unlike an inventory entry).
    pub fn claim_if(self, completion: &Completion) -> Result<io::Result<RegisteredFiles>, Self> {
        if completion.user_data() != self.user_data || completion.ring_id() != self.ring_id {
            return Err(self);
        }
        Ok(completion.result().map(|_| RegisteredFiles {
            base_index: self.base_index,
            count: self.count,
            ring_id: self.ring_id,
        }))
    }
}

/// Per-buffer state a registration keeps for its whole life.
///
/// The registered length is captured at registration time and never
/// recomputed. That matters because `IoBuf::bytes_len` reports the buffer's
/// *current* length, and what the kernel holds is the length it was given --
/// validating a span against the former would let a buffer that had grown
/// address memory the kernel does not know about.
struct BufferState {
    /// Operations outstanding against this one buffer, in either direction.
    outstanding: AtomicUsize,
    /// Of those, the ones the kernel **writes into** this buffer -- that is,
    /// reads. Always a subset of `outstanding`.
    ///
    /// Tracked separately because the two hazards are not the same. Mutating a
    /// buffer races the kernel whichever way the kernel is touching it, so
    /// [`RegisteredBuffers::get_mut`] refuses on `outstanding`. *Reading* a
    /// buffer only races a kernel that is writing into it: a shared read
    /// alongside the kernel's own read of an in-flight write is not a race at
    /// all, and refusing it would deny a sound operation.
    kernel_writes: AtomicUsize,
    /// The length handed to `BuildIoRingRegisterBuffers`, in bytes.
    registered_len: u32,
}

/// Which way the kernel touches a registered buffer for the life of an
/// operation.
///
/// Named from the kernel's point of view rather than the operation's, because
/// that is the direction the data-race question actually turns on, and
/// "a read means the kernel writes" is exactly the inversion that makes this
/// easy to get backwards.
#[derive(Clone, Copy, PartialEq, Eq)]
enum KernelAccess {
    /// A read: the kernel writes into the buffer.
    WritesBuffer,
    /// A write: the kernel reads out of the buffer.
    ReadsBuffer,
}

/// A marker the ring holds while a registered-buffer-indexed read or write
/// is outstanding (M5.2, M5.3).
///
/// Decrements [`RegisteredBuffers`]'s count **for the one buffer the
/// operation addressed** only when the ring returns the entry: an entry the
/// ring forgets rather than returns takes this with it, which correctly
/// leaves the registration believing that use is still outstanding, since
/// nothing proved otherwise.
///
/// The count is per buffer rather than per registration (M13.2) so that
/// [`RegisteredBuffers::get_mut`] can hand out a `&mut` to a *quiet* buffer
/// while operations are still in flight against its neighbours -- which is
/// what an arena a caller refills has to be able to do.
pub struct RegisteredUse {
    state: Arc<[BufferState]>,
    index: usize,
    access: KernelAccess,
}

impl Drop for RegisteredUse {
    fn drop(&mut self) {
        if let Some(state) = self.state.get(self.index) {
            if self.access == KernelAccess::WritesBuffer {
                state.kernel_writes.fetch_sub(1, Ordering::SeqCst);
            }
            state.outstanding.fetch_sub(1, Ordering::SeqCst);
        }
    }
}

/// Buffers registered with a ring so later reads and writes can address them
/// by index instead of handing over a fresh owned buffer each call (M5.2).
///
/// Win32's `IoRing` has no unregister call at all: once registered, an
/// index is valid for the ring's remaining life. This type's own job is
/// narrower than reproducing that -- refusing to let its *backing memory*
/// be freed while this crate still has an operation outstanding against it
/// (M5.3) -- not proving an index will never be touched again by some
/// later, differently-built SQE. That wider hazard belongs to
/// [`crate::IoRing::push_raw`]'s existing `unsafe` contract, the same as any
/// raw use of an index this crate handed out.
pub struct RegisteredBuffers<B: IoBufMut> {
    buffers: ManuallyDrop<Vec<B>>,
    base_index: u32,
    /// One entry per registered buffer, indexed by position within this
    /// registration.
    state: Arc<[BufferState]>,
    ring_id: RingId,
}

impl<B: IoBufMut> RegisteredBuffers<B> {
    /// How many buffers this registration holds.
    #[must_use]
    pub fn len(&self) -> u32 {
        u32::try_from(self.buffers.len()).unwrap_or(u32::MAX)
    }

    /// Whether this registration holds no buffers.
    ///
    /// Always `false` in practice, and a mutation run reports the constant as
    /// surviving for that reason rather than for want of a test. Nothing in
    /// this crate rejects an empty vector, but the kernel refuses the
    /// submission with `E_INVALIDARG`, so a caller never holds an empty
    /// registration. `windows_refuses_an_empty_buffer_registration` pins that
    /// platform behaviour; if a future Windows accepts it, that test fails and
    /// this becomes reachable.
    ///
    /// Kept because it is half of the `len`/`is_empty` pair every Rust
    /// collection surface offers, and because "cannot happen today" is a
    /// weaker claim than "cannot happen".
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.buffers.is_empty()
    }

    /// Borrow the `i`-th buffer's registered bytes, for reading a completed
    /// read's or write's bytes without going through the ring at all.
    ///
    /// # Why it can refuse
    ///
    /// Because otherwise it is a data race with the kernel, reachable from
    /// entirely safe code. [`Batch::read_registered_owned`] takes
    /// `&RegisteredBuffers` and holds only a
    /// [`RegisteredUse`] -- no borrow of this registration outlives the push.
    /// So without the check below, this sequence compiles with no `unsafe`
    /// anywhere and reads memory the kernel is concurrently writing:
    ///
    /// ```ignore
    /// batch.read_registered_owned(&file, &arena, span, (), 0, PushOptions::new())?;
    /// batch.submit()?;                 // kernel now filling that buffer
    /// let bytes = arena.get(span.buffer_index)?;
    /// ```
    ///
    /// The reverse order is the one the check alone could not stop, and is why
    /// this takes `&mut self` -- see below:
    ///
    /// ```ignore
    /// let bytes = arena.get(0)?;       // quiet here, so the check passes
    /// batch.read_registered_owned(&file, &arena, span, (), 0, PushOptions::new())?;
    /// batch.submit()?;                 // kernel now filling that same buffer
    /// let stale = bytes[0];            // ... read through the live borrow
    /// ```
    ///
    /// # Why this takes `&mut self` to hand back a shared slice
    ///
    /// The refusal above is a check at one instant, but the slice it returns
    /// lives as long as the borrow. Taking `&self` made those two different
    /// lengths of time, and the gap was reachable: `read_registered` takes the
    /// registration by *shared* reference, so a caller could take the borrow
    /// while the buffer was quiet, then push and submit a read into that same
    /// buffer and keep reading through the still-live slice. Measured, not
    /// argued -- a probe watched the bytes change under a live `&[u8]` while a
    /// fresh `get` at that same instant correctly refused.
    ///
    /// `&mut self` closes it by making the *borrow* conflict with starting the
    /// operation, not merely its creation: `read_registered(&arena, ..)` cannot
    /// be called while this slice is alive. That costs nothing real, because
    /// **no caller needs to read a buffer during the window it is refused**.
    /// While a read is in flight the bytes are indeterminate -- partially
    /// written, in arbitrary order -- and only become meaningful once the
    /// completion is observed. Earlier or later is always available.
    ///
    /// It does not restrict an arena. An entry holds a [`RegisteredUse`],
    /// not a borrow of this registration, so the shared borrow ends when the
    /// push returns; a caller may still read or fill *quiet* buffers while
    /// operations are outstanding against their neighbours, which is what makes
    /// a registered arena usable at all.
    ///
    /// # Why it refuses less than [`get_mut`] does
    ///
    /// `get_mut` refuses while *any* operation is outstanding, because
    /// mutating races the kernel whichever way the kernel is touching the
    /// buffer. Reading is narrower: it races only a kernel that is **writing
    /// into** the buffer, so this refuses only while a *read* is outstanding --
    /// the bytes of a write in flight are stable, and the kernel is reading
    /// them too.
    ///
    /// A count returns to zero only when a completion is popped against a real
    /// popped [`Completion`], so "not outstanding" means the operation was
    /// *observed* to finish, never merely assumed to have.
    ///
    /// # Why this hands back `&[u8]` and not `&B`
    ///
    /// For the same reason [`get_mut`] does: the slice is exactly the length
    /// handed to `BuildIoRingRegisterBuffers`, which is what the kernel knows
    /// about. `B`'s own `bytes_len()` is a live value that may since have
    /// diverged from it, so answering from the registered length keeps this in
    /// step with the registration rather than with the buffer.
    ///
    /// [`get_mut`]: RegisteredBuffers::get_mut
    ///
    /// # Holding the borrow across a push is refused at compile time
    ///
    /// This is the regression guard for the hazard described above. Taking the
    /// borrow while the buffer is quiet and *then* submitting a read into it
    /// must not compile, and the failure it exists for is the borrow conflict,
    /// `E0502`. That code is pinned below, with a limit worth knowing: rustdoc
    /// checks a pinned code only on a nightly toolchain (or under
    /// `RUSTC_BOOTSTRAP=1`, which CI's `doctest-error-codes` job sets), and on
    /// stable a `compile_fail` example passes on any error. This one went on "passing" after `read_registered` was
    /// removed, because calling a method that does not exist fails to compile
    /// too; the `no_run` twin below, which must compile, is what catches a
    /// rename that reaches both:
    ///
    /// ```compile_fail,E0502
    /// # use windows_ioring_sys::{Batch, IoRing, PushOptions, RegisteredBuffers, RegisteredSpan, SharedHandle};
    /// # fn hazard<B: windows_ioring_sys::IoBufMut>(
    /// #     ring: &mut IoRing,
    /// #     arena: &mut RegisteredBuffers<B>,
    /// #     file: &SharedHandle,
    /// #     span: RegisteredSpan,
    /// # ) -> std::io::Result<u8> {
    /// let bytes: &[u8] = arena.get(0)?;      // quiet here, so the check passes
    /// let mut batch = Batch::new(ring);
    /// batch.read_registered_owned(file, arena, span, (), 0, PushOptions::new())?;
    /// batch.submit()?;                       // kernel now filling that buffer
    /// Ok(bytes[0])                           // ... read through the live borrow
    /// # }
    /// ```
    ///
    /// Reading a *quiet* buffer while an operation is outstanding against a
    /// neighbour is still allowed, because the ring's hold on a registration
    /// is per-buffer -- that is the arena pattern, and it must keep working:
    ///
    /// ```no_run
    /// # use windows_ioring_sys::{Batch, IoRing, PushOptions, RegisteredBuffers, RegisteredSpan, SharedHandle};
    /// # fn arena_still_works<B: windows_ioring_sys::IoBufMut>(
    /// #     ring: &mut IoRing,
    /// #     arena: &mut RegisteredBuffers<B>,
    /// #     file: &SharedHandle,
    /// #     span: RegisteredSpan,
    /// # ) -> std::io::Result<()> {
    /// let mut batch = Batch::new(ring);
    /// batch.read_registered_owned(file, arena, span, (), 0, PushOptions::new())?;
    /// batch.submit()?;
    /// let _neighbour = arena.get(1)?;        // a different, quiet buffer
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// [`io::ErrorKind::InvalidInput`] if `i` is out of range for this
    /// registration, or [`io::ErrorKind::WouldBlock`] if a read is still
    /// outstanding into that buffer -- pop its completion first, which is what
    /// returns the buffer.
    pub fn get(&mut self, i: u32) -> io::Result<&[u8]> {
        let index = usize::try_from(i).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "buffer index does not fit usize",
            )
        })?;
        let Some(state) = self.state.get(index) else {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("buffer index {i} is out of range for this registration"),
            ));
        };
        let kernel_writes = state.kernel_writes.load(Ordering::SeqCst);
        if kernel_writes > 0 {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                format!(
                    "buffer {i} still has {kernel_writes} read(s) outstanding into it; the kernel \
                     may be writing through it. Pop its completion before reading this buffer."
                ),
            ));
        }
        let len = state.registered_len as usize;
        let buffer = self
            .buffers
            .get(index)
            .expect("state and buffers are built together and have equal length");
        // SAFETY: `stable_ptr` is `IoBuf`'s promised stable address for this
        // buffer, and `registered_len` was recorded from that same buffer at
        // registration, so those bytes are allocated and initialized. `&mut
        // self` excludes concurrent mutation through `get_mut` *and* -- because
        // the returned slice borrows it for its whole life -- excludes pushing
        // a new operation against this registration while the slice is alive,
        // since `read_registered` needs `&RegisteredBuffers`. So the
        // zero-`kernel_writes` check above cannot be outrun: no read into this
        // buffer can start before the borrow ends.
        Ok(unsafe { std::slice::from_raw_parts(buffer.stable_ptr(), len) })
    }

    /// Borrow the `i`-th buffer's registered bytes **mutably**, to fill them
    /// before an operation reads out of them (M13.2).
    ///
    /// This is what makes a registered arena usable for writes a caller
    /// composes itself -- a log record, a serialized message -- rather than
    /// only for bytes the kernel put there. Without it, [`get`] can show a
    /// caller what a read landed but nothing can put anything into a buffer
    /// after it is registered.
    ///
    /// # Why this hands back `&mut [u8]` and not `&mut B`
    ///
    /// Because `&mut B` would be **unsound**, and quietly so. Win32's
    /// `IoRing` has no unregister call: the address and length handed to
    /// `BuildIoRingRegisterBuffers` stay live for the ring's remaining life.
    /// A `&mut Vec<u8>` lets safe code assign a whole new vector, `reserve`,
    /// or `resize` -- each of which frees or moves the very allocation the
    /// kernel still holds the address of. No `unsafe` is involved, and it
    /// happens at a moment when nothing is outstanding, so no runtime check
    /// would catch it. [`IoBuf`]'s contract requires that address to stay put
    /// for as long as this crate holds the value, and a registration holds it
    /// forever.
    ///
    /// A byte slice of exactly the registered length grants the one power a
    /// caller actually needs -- writing the bytes -- and none of the powers
    /// that would break the registration.
    ///
    /// # Why it can refuse
    ///
    /// `&mut self` is not on its own enough either. It excludes concurrent
    /// `&self` borrows, but an operation already in flight holds no borrow at
    /// all -- [`Batch::write_registered_owned`] takes `&RegisteredBuffers` for the
    /// length of the call, and the entry it creates keeps only a
    /// [`RegisteredUse`]. So the compiler would happily allow mutation of a
    /// buffer the kernel is reading through right now, which is a data race
    /// with the kernel.
    ///
    /// The runtime half of the check is therefore the per-buffer outstanding
    /// count: this refuses while any operation is still outstanding against
    /// *this* buffer. Neighbouring buffers being in flight is fine, which is
    /// the point of counting per buffer -- an arena a caller refills has to
    /// be able to prepare buffer 5 while buffer 2 is still being written.
    ///
    /// A count returns to zero only when a completion is popped against a
    /// real popped [`Completion`], so "not outstanding" means the operation
    /// was *observed* to finish, never merely assumed to have.
    ///
    /// [`get`]: RegisteredBuffers::get
    ///
    /// # Errors
    ///
    /// [`io::ErrorKind::InvalidInput`] if `i` is out of range for this
    /// registration, or [`io::ErrorKind::WouldBlock`] if an operation is
    /// still outstanding against that buffer -- pop its completion first,
    /// which is what returns the buffer.
    pub fn get_mut(&mut self, i: u32) -> io::Result<&mut [u8]> {
        let index = usize::try_from(i).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "buffer index does not fit usize",
            )
        })?;
        let Some(state) = self.state.get(index) else {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("buffer index {i} is out of range for this registration"),
            ));
        };
        let outstanding = state.outstanding.load(Ordering::SeqCst);
        if outstanding > 0 {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                format!(
                    "buffer {i} still has {outstanding} operation(s) outstanding; the kernel may \
                     be reading or writing through it. Pop its completion before filling this \
                     buffer."
                ),
            ));
        }
        let len = state.registered_len as usize;
        let buffer = self
            .buffers
            .get_mut(index)
            .expect("state and buffers are built together and have equal length");
        // SAFETY: `stable_mut_ptr` is `IoBufMut`'s promised stable, uniquely
        // owned address for this buffer, and `registered_len` was recorded
        // from that same buffer at registration, so those bytes are allocated
        // and initialized. `&mut self` plus the zero-outstanding check above
        // establish that nothing else -- kernel included -- is reading or
        // writing them for this borrow's life.
        Ok(unsafe { std::slice::from_raw_parts_mut(buffer.stable_mut_ptr(), len) })
    }

    /// How many operations are still outstanding against the `i`-th buffer,
    /// or `None` if `i` is out of range.
    ///
    /// Lets a caller pick a quiet buffer without provoking an error from
    /// [`RegisteredBuffers::get_mut`], which is the normal shape for an arena
    /// cycling through its slots.
    #[must_use]
    pub fn outstanding(&self, i: u32) -> Option<usize> {
        self.state
            .get(usize::try_from(i).ok()?)
            .map(|state| state.outstanding.load(Ordering::SeqCst))
    }

    /// Whether `span` lies within this registration -- its buffer index in
    /// range, and `span.offset .. span.offset + span.len` within that buffer's
    /// registered length -- asked without pushing anything. It is the check
    /// every push against a span makes before the kernel sees it, so the
    /// answer is the push's (M34.1).
    ///
    /// For a caller that accepts an operation now and pushes it later, so it
    /// can refuse a span the push would refuse at the moment it accepts it,
    /// rather than at a push nobody is waiting on.
    ///
    /// # Errors
    ///
    /// The [`io::ErrorKind::InvalidInput`] error a push would refuse `span`
    /// with.
    pub fn check_span(&self, span: RegisteredSpan) -> io::Result<()> {
        self.checked_span(span).map(|_| ())
    }

    /// Record that one more operation is outstanding against `span`'s buffer,
    /// and produce the marker whose drop records it finished.
    ///
    /// Call only after [`RegisteredBuffers::checked_span`] has validated
    /// `span`, which is what makes the index below in range.
    fn begin_use(&self, span: RegisteredSpan, access: KernelAccess) -> RegisteredUse {
        let index = span.buffer_index as usize;
        if let Some(state) = self.state.get(index) {
            // The direction-specific counter first, so a concurrent `get` can
            // never observe `outstanding` raised without also seeing the
            // reason it was raised.
            if access == KernelAccess::WritesBuffer {
                state.kernel_writes.fetch_add(1, Ordering::SeqCst);
            }
            state.outstanding.fetch_add(1, Ordering::SeqCst);
        }
        RegisteredUse {
            state: Arc::clone(&self.state),
            index,
            access,
        }
    }

    /// The registered index of the `i`-th buffer, or an
    /// [`io::ErrorKind::InvalidInput`] error if out of range.
    fn checked_index(&self, i: u32) -> io::Result<u32> {
        if i < self.len() {
            Ok(self.base_index + i)
        } else {
            Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("buffer index {i} is out of range for this registration"),
            ))
        }
    }

    /// As [`RegisteredBuffers::checked_index`], plus checking that
    /// `span.offset .. span.offset + span.len` actually fits inside that
    /// buffer -- validating only the index and handing an unchecked range to
    /// the kernel would let it read or write outside the registered
    /// allocation.
    ///
    /// Checked against the length recorded **at registration**, never against
    /// the buffer's current `bytes_len()`. Those are the same number today,
    /// because nothing this type exposes can resize a registered buffer, but
    /// the kernel's view is the registered one and that is what a span has to
    /// fit inside. Validating against a live length would mean a buffer that
    /// somehow grew could address memory the kernel does not know about.
    fn checked_span(&self, span: RegisteredSpan) -> io::Result<u32> {
        let index = self.checked_index(span.buffer_index)?;
        let registered_len = self
            .state
            .get(span.buffer_index as usize)
            .expect("checked_index just validated this index")
            .registered_len;
        let fits = u64::from(span.offset)
            .checked_add(u64::from(span.len))
            .is_some_and(|end| end <= u64::from(registered_len));
        if !fits {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!(
                    "span offset {} + len {} does not fit inside buffer {} \
                     ({registered_len} registered bytes)",
                    span.offset, span.len, span.buffer_index
                ),
            ));
        }
        Ok(index)
    }
}

impl<B: IoBufMut> Drop for RegisteredBuffers<B> {
    fn drop(&mut self) {
        if self
            .state
            .iter()
            .any(|state| state.outstanding.load(Ordering::SeqCst) > 0)
        {
            // Refused, not silently permitted (M5.3): freeing now would
            // leave an outstanding `IORING_BUFFER_REF` pointing at freed
            // memory. Loud in debug builds; in release, leaking is the safe
            // failure mode -- the same choice the ring makes for its inventory
            // when rundown cannot prove quiescence (D-73): "leak is safe,
            // use-after-free is not". So `buffers` is
            // simply never reclaimed rather than freed out from under a
            // still-outstanding op. Any one buffer still in use holds the
            // whole registration, because they are freed together.
            //
            // Silent while already panicking (M23.4). A second panic during
            // unwind aborts the process, which *replaces* the failure that
            // started the unwind: a test asserting on a leaked slot reports
            // `STATUS_STACK_BUFFER_OVERRUN` instead of its own message, so
            // the diagnosis is destroyed by the guard meant to aid it. The
            // leak is still refused -- the early `return` below is what makes
            // leaking rather than freeing the failure mode, and it happens
            // either way.
            debug_assert!(
                std::thread::panicking(),
                "RegisteredBuffers dropped while an operation still references it"
            );
            return;
        }
        // SAFETY: every per-buffer count is zero, so no operation still
        // references these buffers, and this is the only place `buffers` is
        // ever dropped.
        unsafe { ManuallyDrop::drop(&mut self.buffers) };
    }
}

/// A [`Batch::register_buffers`] push not yet matched to its completion.
///
/// `buffers` is `ManuallyDrop` and this type's own `Drop` is deliberately
/// empty, mirroring the inventory (PR #20 review response): the registration is
/// already queued via `BuildIoRingRegisterBuffers` the instant
/// [`Batch::register_buffers`] returns, before any completion is observed,
/// so a caller that drops this without ever matching a completion has no
/// proof the kernel is done deciding whether to retain these addresses.
/// Freeing them here anyway would risk handing memory the kernel still
/// references back to the allocator; leaking is the safe failure mode, the
/// same choice the ring's inventory (D-73) and [`RegisteredBuffers`] both
/// already make.
pub struct PendingBufferRegistration<B: IoBufMut> {
    user_data: usize,
    base_index: u32,
    buffers: ManuallyDrop<Vec<B>>,
    /// The lengths handed to `BuildIoRingRegisterBuffers`, captured there
    /// rather than recomputed here, so what a span is later validated against
    /// is exactly what the kernel was told.
    registered_lens: Vec<u32>,
    ring_id: RingId,
}

impl<B: IoBufMut> PendingBufferRegistration<B> {
    /// This push's `UserData` identity, to match against a popped
    /// [`Completion`].
    #[must_use]
    pub fn user_data(&self) -> usize {
        self.user_data
    }

    /// Turn this pending registration into usable [`RegisteredBuffers`] once
    /// `completion` names it, or hand it back unchanged if `completion`
    /// names a different operation.
    ///
    /// # Errors
    ///
    /// The inner `Result` is `Err` if the registration itself failed; the
    /// buffers are dropped normally in that case, exactly as if they had
    /// never been registered -- a matched completion, success or failure, is
    /// exactly the proof this type's `Drop` is waiting for.
    pub fn claim_if(
        mut self,
        completion: &Completion,
    ) -> Result<io::Result<RegisteredBuffers<B>>, Self> {
        if completion.user_data() != self.user_data || completion.ring_id() != self.ring_id {
            return Err(self);
        }
        // SAFETY: `completion` names this exact registration, so the kernel
        // has already decided whether to retain these addresses -- the
        // condition this type's `Drop` would otherwise wait forever for.
        let buffers = unsafe { ManuallyDrop::take(&mut self.buffers) };
        let base_index = self.base_index;
        let ring_id = self.ring_id;
        let registered_lens = std::mem::take(&mut self.registered_lens);
        Ok(completion.result().map(move |_| {
            let state: Arc<[BufferState]> = registered_lens
                .into_iter()
                .map(|registered_len| BufferState {
                    outstanding: AtomicUsize::new(0),
                    kernel_writes: AtomicUsize::new(0),
                    registered_len,
                })
                .collect();
            debug_assert_eq!(
                state.len(),
                buffers.len(),
                "one state entry per registered buffer"
            );
            RegisteredBuffers {
                buffers: ManuallyDrop::new(buffers),
                base_index,
                state,
                ring_id,
            }
        }))
    }
}

impl<B: IoBufMut> Drop for PendingBufferRegistration<B> {
    fn drop(&mut self) {
        // Deliberately empty; see this type's own doc comment.
    }
}

impl<B: IoBufMut> std::fmt::Debug for PendingBufferRegistration<B> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Not derived: deriving would require `B: Debug`, which a caller's
        // buffer type need not satisfy.
        f.debug_struct("PendingBufferRegistration")
            .field("user_data", &self.user_data)
            .field("base_index", &self.base_index)
            .finish_non_exhaustive()
    }
}

/// A scoped, exclusive submission window over one [`IoRing`] (D-5).
///
/// Holding `&mut IoRing` for its lifetime is what turns Win32's own
/// "you must serialize submission" footnote into a compiler-enforced
/// guarantee: two batches over the same ring cannot coexist. Every push
/// queues its SQE the instant it succeeds -- there is no rewind -- so
/// `Batch` submits on [`Drop`] rather than leaving queued SQEs for some
/// later, unrelated submit to discover (D-5).
pub struct Batch<'ring, T = (), X = (), K = Open> {
    ring: &'ring mut IoRing<T, X, K>,
    submitted: bool,
}

impl<'ring, T, X, K> Batch<'ring, T, X, K> {
    /// Open a batch over `ring`. The batch has the ring's kind: over an [`Owned`](crate::Owned) ring it has
    /// no call that queues an operation the ring holds nothing for ([D-91](../DESIGN-NOTES.md#d-91)).
    pub fn new(ring: &'ring mut IoRing<T, X, K>) -> Self {
        Self {
            ring,
            submitted: false,
        }
    }

    fn require(&self, op: Op) -> io::Result<()> {
        if self.ring.supports(op) {
            Ok(())
        } else {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                format!("this ring does not support {op:?}"),
            ))
        }
    }

    /// Reserve an identity for an inventory push.
    ///
    /// Split from [`Batch::finish_owned`] because the identity has to exist
    /// *before* the SQE is built -- the kernel echoes it back -- while what
    /// the ring holds is only decided once the build has succeeded.
    fn begin_owned(&mut self) -> io::Result<(usize, OperationId)> {
        let user_data = self.ring.accounting_mut().reserve_user_data()?;
        Ok((user_data, OperationId::new(user_data, self.ring.ring_id())))
    }

    /// The shared tail of every inventory push: stow on success; on failure,
    /// release the reservation, release what the ring held for the operation
    /// (the file guard and any registered-buffer use), and hand the payload
    /// and sidecar back in a [`PushRefused`] ([D-80](../DESIGN-NOTES.md#d-80)).
    ///
    /// Handing back is sound for the same reason dropping was before it: a
    /// `Build*` that fails queued no SQE, so the kernel never saw the address
    /// and nothing will ever complete against it. Leaking is correct only once
    /// the kernel *has* seen an address, which a failed build rules out.
    fn finish_owned(
        &mut self,
        hr: windows_sys::core::HRESULT,
        id: OperationId,
        payload: Option<T>,
        extra: X,
        held: Held,
    ) -> Result<OperationId, PushRefused<T, X>> {
        match check(hr) {
            Ok(()) => {
                self.ring.stow(
                    id,
                    Entry {
                        payload,
                        extra,
                        held,
                    },
                );
                Ok(id)
            }
            Err(error) => {
                self.ring.cancel_reservation(id.user_data());
                drop(held);
                Err(PushRefused::new(error, payload, extra))
            }
        }
    }

    /// Refuse a [`RegisteredBuffers`] that did not come from this batch's own
    /// ring (PR #20 review response): its index space is only meaningful
    /// against the ring that registered it, and a different ring may have an
    /// entirely unrelated (or already-freed) buffer at the same index.
    fn check_registration_ring<B: IoBufMut>(
        &self,
        registration: &RegisteredBuffers<B>,
    ) -> io::Result<()> {
        if registration.ring_id == self.ring.ring_id() {
            Ok(())
        } else {
            Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "this RegisteredBuffers was registered on a different IoRing",
            ))
        }
    }

    /// Submit everything queued so far, returning the number of entries the
    /// kernel accepted.
    ///
    /// That count is entries *submitted*, never entries completed (M10.2):
    /// draining is [`crate::IoRing::try_pop`]'s job and stays the caller's.
    ///
    /// # Errors
    ///
    /// Returns any error from `SubmitIoRing`. **An error means the entries
    /// were not submitted and remain in the submission queue**, which is the
    /// documented behaviour of that call: *"If this function returns an error
    /// other than IORING_E_WAIT_TIMEOUT, then all entries remain in the
    /// submission queue."* They are not lost and they are not rewound
    /// ([D-5](../DESIGN-NOTES.md#d-5)) -- a later submit on this ring is what
    /// runs them, so **the buffers they reference must stay alive**. Whether
    /// and when to submit again is the caller's policy, not this crate's.
    pub fn submit(self) -> io::Result<u32> {
        self.submit_and_wait(0, 0)
    }

    /// Close the batch **without submitting** what it queued: the entries stay in the submission
    /// queue, held by the ring with their buffers, until a later [`Batch::submit`] -- over a batch
    /// that queues nothing, if need be -- issues them ([D-90](../DESIGN-NOTES.md#d-90)).
    ///
    /// A batch submits as it drops by default ([D-5](../DESIGN-NOTES.md#d-5)), so that nothing is
    /// left queued for a later, unrelated submit to discover. This is the caller's explicit choice to
    /// own the doorbell instead, to ring it once for several pushes, or to learn that it failed:
    /// the drop's own submit reports nothing. It is the state a failed [`Batch::submit`] leaves,
    /// reached on purpose. **Until the submit, the kernel has not seen the entries**, so none will
    /// complete.
    pub fn leave_queued(mut self) {
        self.submitted = true;
    }

    /// Submit everything queued so far, then block until at least
    /// `wait_operations` of them have completed or `timeout_ms` elapses
    /// (D-3's Model B primitive: the fused submit-and-wait *is* the event
    /// loop for a pinned-thread consumer).
    ///
    /// Returning does not mean `wait_operations` completions are poppable:
    /// the timeout can expire first, and the returned count is entries
    /// submitted rather than completed (M10.2). Drain with
    /// [`crate::IoRing::try_pop`] and count for yourself.
    ///
    /// **A wait that expires is success, not failure** (M26.8). `SubmitIoRing`
    /// answers `IORING_E_WAIT_TIMEOUT` in that case, and its documented
    /// meaning is *"All operations were submitted without error and the
    /// subsequent wait timed out"* -- so the submission half did everything it
    /// was asked to. Reporting that as an `Err` is the defect `M21.6` fixed
    /// for [`crate::IoRing::pop_within`]; this site was missed by that sweep
    /// and kept it until `M26.8`. The distinction is not cosmetic: an `Err`
    /// here means the entries are **still queued**, and a caller who frees
    /// their buffers on seeing one would hand the kernel freed memory on the
    /// next submit.
    ///
    /// # Errors
    ///
    /// Returns any error from `SubmitIoRing` **other than an expired wait**.
    /// As [`Batch::submit`], an error means the entries remain in the
    /// submission queue and their buffers must stay alive.
    pub fn submit_and_wait(mut self, wait_operations: u32, timeout_ms: u32) -> io::Result<u32> {
        self.do_submit(wait_operations, timeout_ms)
    }

    fn do_submit(&mut self, wait_operations: u32, timeout_ms: u32) -> io::Result<u32> {
        let mut submitted = 0_u32;
        // SAFETY: `self.ring`'s handle is live.
        let hr = unsafe {
            crate::sys::submit(
                self.ring.ring_id(),
                self.ring.raw_handle(),
                wait_operations,
                timeout_ms,
                &raw mut submitted,
            )
        };
        // Marked attempted *before* propagating a failure (PR #20 review
        // response): `submit_and_wait` takes `self` by value, so this call
        // failing still runs `Drop` once its caller's `self` goes out of
        // scope. Setting this first is what stops `Drop` from silently
        // retrying an explicit submit that already ran -- which could
        // succeed on the retry, submitting operations the caller's `Err`
        // never told them about.
        self.submitted = true;
        // `IORING_E_WAIT_TIMEOUT` is not a submission failure: the documented
        // meaning of that code is that every entry went in and only the wait
        // ran out (M26.8). Passing it to `check` reported a successful submit
        // as an error, which is `M21.6`'s defect at the one site that sweep
        // did not reach -- and the worse half is that it made an `Err` here
        // ambiguous between "still queued" and "submitted, wait expired",
        // which have opposite consequences for buffer ownership.
        if !crate::ring::every_entry_was_submitted(hr) {
            check(hr)?;
        }
        Ok(submitted)
    }
}
impl<T, X, K> Drop for Batch<'_, T, X, K> {
    fn drop(&mut self) {
        if !self.submitted {
            // Best-effort: Drop cannot propagate an error, and a batch that
            // queued nothing submits zero entries harmlessly (D-5).
            let _ = self.do_submit(0, 0);
        }
    }
}

// The push surface lives in `push.rs`. It is a child rather than a sibling so
// it can see this module's private plumbing -- `require`, `begin_owned`,
// `finish_owned`, `check_registration_ring`, `handle_ref`, `checked_len` and
// the private `sqe_flags`/`raw` accessors on the option types. No public path
// changes: an inherent `impl` is found by the type, not by the module it was
// written in.
mod push;

#[cfg(test)]
mod tests;