windows-ioring-sys 0.7.2

// Copyright (c) 2026 Mike Grier
use super::{Completion, InjectedFailure, IoRing, Op, OpSupport};
use crate::IoRingErrorExt;
use crate::capability::{RingVersion, capabilities};

#[test]
fn op_support_starts_empty() {
    assert!(!OpSupport::default().contains(Op::Read));
    assert!(!OpSupport::default().contains(Op::Nop));
}

#[test]
fn every_named_version_the_host_supports_creates_and_closes() {
    let caps = capabilities().expect("capabilities");
    let mut created_at_least_one = false;
    for version in [RingVersion::V1, RingVersion::V2, RingVersion::V3] {
        if version > caps.max_version {
            continue;
        }
        let ring = IoRing::with_version(version, 64, 128).expect("create at a supported version");
        assert_eq!(ring.version(), version);
        drop(ring);
        created_at_least_one = true;
    }
    assert!(
        created_at_least_one,
        "the host should support at least IORING_VERSION_1"
    );
}

#[test]
fn creating_a_ring_above_the_hosts_maximum_version_fails() {
    let caps = capabilities().expect("capabilities");
    let too_high = RingVersion::from_raw(caps.max_version.raw() + 1);
    let error =
        IoRing::with_version(too_high, 64, 128).expect_err("an unsupported version must fail");
    assert_eq!(error.kind(), std::io::ErrorKind::Other);
}

#[test]
fn capability_reporting_never_claims_more_than_is_io_ring_op_supported_reports() {
    let ring = IoRing::new(64, 128).expect("create ring");
    for op in [
        Op::Nop,
        Op::Read,
        Op::Write,
        Op::Flush,
        Op::RegisterFiles,
        Op::RegisterBuffers,
        Op::Cancel,
    ] {
        assert_eq!(
            ring.supports(op),
            ring.supports_raw(op.code()),
            "cached support for {op:?} disagrees with a direct IsIoRingOpSupported call"
        );
    }
}

#[test]
fn supports_reports_exactly_the_capability_set_it_was_given() {
    // `IoRing::supports -> true` survived: every op named in this crate is
    // genuinely supported on any real host these tests run on, so the honest
    // answer and the constant agree everywhere a test could ask a real ring.
    // `set_supported_ops_for_test` constructs the disagreement instead of
    // hoping to find a host that lacks something.
    let mut ring = IoRing::new(8, 8).expect("create ring");
    ring.set_supported_ops_for_test(&[Op::Nop, Op::Read]);

    assert!(ring.supports(Op::Nop));
    assert!(ring.supports(Op::Read));
    assert!(
        !ring.supports(Op::Write),
        "an op left out of the constructed set must read back as unsupported"
    );
    assert!(!ring.supports(Op::Cancel));
}

#[test]
fn nop_read_and_write_are_supported_on_any_real_ring() {
    // A sanity floor: every documented IoRing version supports at least
    // these three. If this ever fails, either the host is exotic enough to
    // need investigating, or the probe itself is broken.
    let ring = IoRing::new(64, 128).expect("create ring");
    assert!(ring.supports(Op::Nop));
    assert!(ring.supports(Op::Read));
    assert!(ring.supports(Op::Write));
}

// --- outstanding-operation accounting and rundown (M2.4) ---

#[test]
fn run_down_returns_once_a_recorded_completion_zeroes_the_count() {
    let mut ring = IoRing::new(64, 128).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    assert_eq!(ring.outstanding(), 1);
    // Recording the completion up front proves run_down rechecks the count
    // rather than always performing at least one wait: it must return
    // without ever calling SubmitIoRing, or this test would hang for
    // RUN_DOWN_POLL_MS waiting on a completion that was never real.
    ring.record_completion(op);
    ring.run_down()
        .expect("run_down with the count already settled");
    assert_eq!(ring.outstanding(), 0);
}

#[test]
fn dropping_a_ring_actually_runs_its_drop_body() {
    // `<impl Drop for IoRing>::drop -> ()` survived: nothing distinguished a
    // ring that ran rundown-and-close from one that silently leaked its
    // kernel handle, because closing it is invisible to every test that only
    // asks the ring itself. `DROP_RUNS` is incremented as the first line of
    // the real body, so a mutation that replaces the whole body removes the
    // increment along with everything else.
    //
    // The counter is thread-local, and an EXACT count is asserted. An earlier
    // version used a process-wide static and asserted `after > before`, which
    // is not race-free: `cargo test` runs tests as threads in one process, so
    // another test dropping any ring between the two reads satisfies the
    // assertion on its own -- masking precisely the mutant this exists to
    // catch. Only rings dropped on this thread can move a thread-local, and
    // this test drops exactly one.
    let before = super::DROP_RUNS.with(std::cell::Cell::get);
    let ring = IoRing::new(8, 8).expect("create ring");
    drop(ring);
    let after = super::DROP_RUNS.with(std::cell::Cell::get);
    assert_eq!(
        after,
        before + 1,
        "dropping one ring must run its Drop impl exactly once (before={before}, after={after})"
    );
}

/// `IoRing::drop`'s close assert, traversed for real (M23.5).
///
/// M23.4 narrowed both asserts in `IoRing::drop` to fire only outside an
/// unwind, and measured that suppressing them entirely left every test in the
/// crate green -- so the guards were unverified in the direction that matters.
/// The item that followed assumed reaching them required a fault-injection
/// seam over every raw HRESULT the ring's Win32 calls return, and priced that
/// seam's blast radius. It is not required: the kernel already refuses a null
/// ring handle, and `ring::tests` is a child of `ring`, so it can build an
/// `IoRing` around one. See [`IoRing::refused_by_the_kernel`] for why null
/// specifically, and why a non-null stand-in would crash instead.
// Debug-only, as is the rundown test below: both expect a `debug_assert!` in
// `IoRing::drop`, which a release build compiles out. Gated together with
// `batch::tests::dropping_a_registration_with_work_outstanding_is_refused`,
// the third test of that shape.
#[cfg(debug_assertions)]
#[test]
#[should_panic(expected = "CloseIoRing failed")]
fn a_ring_whose_close_the_kernel_refuses_reports_the_close_failure() {
    let ring = IoRing::refused_by_the_kernel();

    // Nothing is outstanding, so `run_down` returns `Ok` without ever
    // submitting. That is what makes the close assert the only one this test
    // can reach -- see the sibling test for the other one.
    assert_eq!(
        ring.accounting.outstanding(),
        0,
        "a fresh ledger has nothing outstanding"
    );
    drop(ring);
}

/// `IoRing::drop`'s rundown assert, traversed for real (M23.5).
///
/// The same construction as the sibling test above, plus the one thing that
/// selects the other assert: `run_down` submits only *while* something is
/// outstanding, so a ring with an empty ledger never calls `SubmitIoRing` and
/// never fails. Reserving one identity makes the loop run once, that submit
/// is refused, and `run_down` returns the error the assert names.
///
/// The two `expected` strings are what keep these tests honest about which
/// assert they reached: if this one fell through to the close instead, its
/// panic would say `CloseIoRing failed` and the test would go red rather than
/// pass for the wrong reason. That is sabotaged in [sabotage.json], not merely
/// asserted here.
///
/// [sabotage.json]: ../../sabotage.json
#[cfg(debug_assertions)]
#[test]
#[should_panic(expected = "IoRing rundown failed before close")]
fn a_ring_whose_rundown_the_kernel_refuses_reports_the_rundown_failure() {
    let mut ring = IoRing::refused_by_the_kernel();

    ring.accounting
        .reserve_user_data()
        .expect("a fresh ring's identity space is not exhausted");
    assert_eq!(
        ring.accounting.outstanding(),
        1,
        "rundown must have a reason to submit, or it cannot fail"
    );

    drop(ring);
}

// --- The registration arrays outlive an unproven rundown ---
//
// The kernel reads both registration arrays -- `IORING_BUFFER_INFO` and the
// file-handle array -- when the registration op runs (D-32), so `Drop` may free
// them only on the proof the inventory needs: rundown showed nothing is
// outstanding. PR #113 review found the buffer array freed on every path,
// because it was an ordinary field; the handle array was not held at all.
// These pin both halves of the fix, for both arrays.

/// One descriptor, pointing at nothing the kernel is ever shown: no SQE is
/// built from it, so it is only a non-empty array for `Drop` to decide about.
fn one_buffer_info() -> Vec<windows_sys::Win32::Storage::FileSystem::IORING_BUFFER_INFO> {
    vec![
        windows_sys::Win32::Storage::FileSystem::IORING_BUFFER_INFO {
            Address: std::ptr::null_mut(),
            Length: 0,
        },
    ]
}

/// One null handle, for the same reason: nothing is ever built from it.
fn one_file_handle() -> Vec<*mut std::ffi::c_void> {
    vec![std::ptr::null_mut()]
}

/// Holds a non-empty array of each kind, as a ring that registered both would.
fn hold_both(ring: &mut IoRing) {
    let _ = ring.hold_registered_buffer_infos(one_buffer_info());
    let _ = ring.hold_registered_file_handles(one_file_handle());
}

/// The compiler's own drop of the field is what freed the buffer array on the
/// failed-rundown path -- and in a debug build it did so during the unwind out
/// of `Drop`'s assert, where no line of the body can intervene. Only the
/// field's type prevents that, so the type is what this pins: it does not
/// compile if the field stops being `ManuallyDrop`. Entirely compile-time, so
/// it opens no ring (D-49).
#[test]
fn the_registration_arrays_are_never_dropped_by_the_compiler() {
    fn pinned(ring: &IoRing) -> &std::mem::ManuallyDrop<super::LateReadArrays> {
        &ring.late_read
    }
    let _ = pinned;
}

/// The accepting direction: a ring whose rundown succeeds releases both
/// arrays. Without this, a `Drop` that leaked them on every path would pass
/// the test below and leak an allocation per registration on every ring.
///
/// No kernel ring is needed (D-49): the release turns on rundown alone, and
/// with nothing outstanding rundown returns without submitting. The null
/// handle's close is then refused, which is why the ring is dropped during an
/// unwind -- the close assert stands down there, as in the refusing test below.
#[test]
fn a_quiesced_ring_releases_its_registration_arrays() {
    let before = super::LATE_READ_ARRAYS_RELEASED.with(std::cell::Cell::get);

    let unwound = std::panic::catch_unwind(|| {
        let mut ring = IoRing::refused_by_the_kernel();
        hold_both(&mut ring);
        assert_eq!(ring.accounting.outstanding(), 0, "nothing is outstanding");
        panic!("deliberate: drop the ring during an unwind");
    });
    assert!(unwound.is_err(), "the closure must have unwound");

    let after = super::LATE_READ_ARRAYS_RELEASED.with(std::cell::Cell::get);
    assert_eq!(
        after,
        before + 2,
        "a ring with nothing outstanding must release both registration arrays"
    );
}

/// The refusing direction: a ring whose rundown fails keeps both arrays.
///
/// Dropped during an unwind, because that is the only way to reach the release
/// decision in a debug build: outside one, the rundown assert panics first and
/// the decision is never made, so a `Drop` that released unconditionally
/// would pass. During an unwind both asserts stand down and the body runs to
/// the end, in either profile.
#[test]
fn a_ring_whose_rundown_fails_keeps_its_registration_arrays() {
    let before = super::LATE_READ_ARRAYS_RELEASED.with(std::cell::Cell::get);

    let unwound = std::panic::catch_unwind(|| {
        let mut ring = IoRing::refused_by_the_kernel();
        hold_both(&mut ring);
        ring.accounting
            .reserve_user_data()
            .expect("a fresh ring's identity space is not exhausted");
        panic!("deliberate: drop the ring during an unwind");
    });
    assert!(unwound.is_err(), "the closure must have unwound");

    let after = super::LATE_READ_ARRAYS_RELEASED.with(std::cell::Cell::get);
    assert_eq!(
        after, before,
        "a ring that could not prove quiescence must not release an array the kernel may still read"
    );
}

// --- The registered files outlive the ring (M30.2, D-81) ---
//
// `Batch::register_shared_files` hands the ring files whose handles the kernel
// may reference for the ring's whole life. `Drop` releases them only on a
// quiesced rundown, after `CloseIoRing`, and leaks them otherwise. The two
// tests below use the null-handle ring, whose close the kernel refuses: the
// release must not depend on the close succeeding, or a quiesced ring with a
// failed close would leak every file it registered.

/// Two real file handles, to stand for a registration's files. No SQE is ever
/// built from them; they are only something for `Drop` to decide about.
fn two_shared_files() -> Vec<win_shared_os_owned_handle::SharedHandle> {
    let exe = std::env::current_exe().expect("the test binary's path");
    (0..2)
        .map(|_| {
            let file = std::fs::File::open(&exe).expect("open the test binary for read");
            win_shared_os_owned_handle::SharedHandle::new(std::os::windows::io::OwnedHandle::from(
                file,
            ))
        })
        .collect()
}

/// As [`the_registration_arrays_are_never_dropped_by_the_compiler`]: only the
/// field's type stops the compiler freeing the files on the failed-rundown
/// path, so the type is what this pins.
#[test]
fn the_registered_files_are_never_dropped_by_the_compiler() {
    fn pinned(
        ring: &IoRing,
    ) -> &std::mem::ManuallyDrop<Vec<win_shared_os_owned_handle::SharedHandle>> {
        &ring.registered_files
    }
    let _ = pinned;
}

/// The accepting direction: a quiesced ring releases every registered file,
/// even when its close is refused.
#[test]
fn a_quiesced_ring_releases_its_registered_files() {
    let before = super::REGISTERED_FILES_RELEASED.with(std::cell::Cell::get);

    let unwound = std::panic::catch_unwind(|| {
        let mut ring = IoRing::refused_by_the_kernel();
        ring.hold_registered_shared_files(two_shared_files());
        assert_eq!(ring.accounting.outstanding(), 0, "nothing is outstanding");
        panic!("deliberate: drop the ring during an unwind");
    });
    assert!(unwound.is_err(), "the closure must have unwound");

    let after = super::REGISTERED_FILES_RELEASED.with(std::cell::Cell::get);
    assert_eq!(
        after,
        before + 2,
        "a ring with nothing outstanding must release every registered file"
    );
}

/// The refusing direction: a ring whose rundown fails keeps its registered
/// files, so a handle the kernel may still use is never closed under it.
/// Dropped during an unwind for the reason given on
/// [`a_ring_whose_rundown_fails_keeps_its_registration_arrays`].
#[test]
fn a_ring_whose_rundown_fails_keeps_its_registered_files() {
    let before = super::REGISTERED_FILES_RELEASED.with(std::cell::Cell::get);

    let unwound = std::panic::catch_unwind(|| {
        let mut ring = IoRing::refused_by_the_kernel();
        ring.hold_registered_shared_files(two_shared_files());
        ring.accounting
            .reserve_user_data()
            .expect("a fresh ring's identity space is not exhausted");
        panic!("deliberate: drop the ring during an unwind");
    });
    assert!(unwound.is_err(), "the closure must have unwound");

    let after = super::REGISTERED_FILES_RELEASED.with(std::cell::Cell::get);
    assert_eq!(
        after, before,
        "a ring that could not prove quiescence must not close a handle the kernel may still use"
    );
}

// --- The fault-injection seam (M16.3) ---

/// A real completion for a real, finished operation.
///
/// Every test below starts from one of these rather than from
/// `Completion::synthetic`, because the point of the seam is that it transforms
/// something the ring genuinely popped. Starting from a fabrication would test
/// a different thing entirely.
fn real_completion(ring: &mut IoRing) -> (usize, crate::Completion) {
    use crate::{Batch, FlushCoverage, FlushMode};
    use std::os::windows::io::AsRawHandle;

    let path = std::env::temp_dir().join(format!(
        "windows-ioring-sys-injection-{}-{:?}.tmp",
        std::process::id(),
        std::thread::current().id()
    ));
    std::fs::write(&path, b"x").expect("create fixture");
    let file = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(&path)
        .expect("open fixture");

    let mut batch = Batch::new(ring);
    // SAFETY: `file` outlives this call and the drain below.
    let user_data = unsafe {
        batch.flush_raw(
            file.as_raw_handle(),
            FlushCoverage::Unordered,
            FlushMode::Default,
        )
    }
    .expect("queue a flush");
    batch.submit_and_wait(1, 30_000).expect("submit and wait");

    let completion = super::pop_within(ring, "the flush's completion");
    let _ = std::fs::remove_file(&path);
    (user_data, completion)
}

#[test]
fn an_injected_failure_replaces_a_real_success() {
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let (user_data, completion) = real_completion(&mut ring);
    completion
        .result()
        .expect("the flush really did succeed, or this test proves nothing");

    let injected = completion.with_injected_failure(crate::InjectedFailure::Win32(
        windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED,
    ));
    let error = injected.result().expect_err("the injected failure applies");
    assert_eq!(
        (crate::IoRingErrorExt::as_ioring_error(&error)
            .expect("an IoRingError")
            .code() as u32)
            & 0xFFFF,
        windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED,
        "the Win32 code must survive the HRESULT wrapping the kernel would apply"
    );
    assert_eq!(
        injected.user_data(),
        user_data,
        "injection must not disturb the operation's identity"
    );
}

#[test]
fn an_injected_failure_preserves_the_identity_a_token_claims_against() {
    // The safety-critical property, and the reason the seam is worth having:
    // injection transforms a completion for an operation that *genuinely
    // finished*, so claiming against the result stays exactly as sound as
    // claiming against the original -- and must still work, or the seam could
    // not test the claim paths that failure handling lives on.
    //
    // A real, payload-carrying read throughout: nothing here is fabricated.
    use crate::{Batch, PushOptions};
    use std::os::windows::io::AsRawHandle;

    let path = std::env::temp_dir().join(format!(
        "windows-ioring-sys-injection-claim-{}.tmp",
        std::process::id()
    ));
    std::fs::write(&path, b"hello").expect("create fixture");
    let file = std::fs::OpenOptions::new()
        .read(true)
        .open(&path)
        .expect("open fixture");

    let mut ring = IoRing::<Vec<u8>>::with_inventory(16, 16).expect("create ring");
    let mut batch = Batch::new(&mut ring);
    // SAFETY: `file` outlives the operation; the ring holds the buffer until
    // the pop below hands it back.
    unsafe {
        batch.read_raw_owned(
            file.as_raw_handle(),
            vec![0_u8; 5],
            (),
            0,
            PushOptions::new(),
        )
    }
    .expect("queue a read");
    batch.submit_and_wait(1, 30_000).expect("submit and wait");
    let (completion, held) = ring
        .pop_within(std::time::Duration::from_secs(30))
        .expect("pop")
        .expect("the read's completion");
    completion
        .result()
        .expect("the read really did succeed, or this test proves nothing");

    let injected = completion
        .with_injected_failure(crate::InjectedFailure::Ring(crate::RingCondition::Corrupt));
    assert!(injected.result().is_err(), "the injected failure applies");

    let buffer = held
        .map(|(payload, ())| payload.expect("a read carries a buffer"))
        .expect("a failed completion still hands back its own payload");
    assert_eq!(
        buffer, b"hello",
        "claiming a failed operation must still hand the buffer back -- that is \
         exactly what stops a failure path from leaking it, which is the defect \
         this seam exists to let a test reach"
    );

    let _ = std::fs::remove_file(&path);
}

#[test]
fn an_injected_failure_zeroes_the_transferred_byte_count() {
    // The deletion of `information: 0,` from the struct-update survived: with
    // it gone, `..self` supplies the *original* transfer count, so an
    // injected "failure" completion silently keeps reporting real bytes
    // transferred. `Completion::result` cannot show this -- it only returns
    // `information` on success, and this seam only injects failure -- so the
    // field is read directly. This module is `ring.rs`'s own child and can
    // see it, which is exactly what an earlier version of this file's comment
    // (just above) said was impossible.
    use crate::{Batch, PushOptions};
    use std::os::windows::io::AsRawHandle;

    let path = std::env::temp_dir().join(format!(
        "windows-ioring-sys-injection-information-{}-{:?}.tmp",
        std::process::id(),
        std::thread::current().id()
    ));
    std::fs::write(&path, b"hello").expect("create fixture");
    let file = std::fs::OpenOptions::new()
        .read(true)
        .open(&path)
        .expect("open fixture");

    let mut ring = IoRing::<Vec<u8>>::with_inventory(16, 16).expect("create ring");
    let mut batch = Batch::new(&mut ring);
    // SAFETY: `file` outlives the operation, and the completion is popped
    // below before it is dropped.
    unsafe {
        batch.read_raw_owned(
            file.as_raw_handle(),
            vec![0_u8; 5],
            (),
            0,
            PushOptions::new(),
        )
    }
    .expect("queue a read");
    batch.submit_and_wait(1, 30_000).expect("submit and wait");
    let completion = super::pop_within(&mut ring, "the fixture read's completion");
    assert_eq!(
        completion.information, 5,
        "the fixture must transfer five real bytes, or this test proves nothing"
    );

    let injected = completion
        .with_injected_failure(crate::InjectedFailure::Ring(crate::RingCondition::Corrupt));
    assert_eq!(
        injected.information, 0,
        "an injected failure must report zero transferred, not the real completion's count"
    );

    let _ = std::fs::remove_file(&path);
}

#[test]
fn each_spelling_of_a_failure_produces_the_condition_it_names() {
    // The three variants exist so a call site reads as what it is testing.
    // If they did not agree with the codes they name, that legibility would be
    // a lie.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let (_, completion) = real_completion(&mut ring);

    let ring_failure = completion.with_injected_failure(crate::InjectedFailure::Ring(
        crate::RingCondition::SubmissionQueueFull,
    ));
    let error = ring_failure.result().expect_err("fails");
    assert!(
        crate::IoRingErrorExt::is_submission_queue_full(&error),
        "a Ring(..) injection must be recognised by the named predicate"
    );

    let raw = crate::RingCondition::Corrupt.code();
    let hresult_failure = completion.with_injected_failure(crate::InjectedFailure::Hresult(raw));
    assert_eq!(
        crate::IoRingErrorExt::ring_condition(&hresult_failure.result().expect_err("fails")),
        Some(crate::RingCondition::Corrupt),
        "a raw HRESULT injection must resolve to the same condition"
    );
}

#[test]
#[should_panic(expected = "injects failure only, never success")]
fn injecting_a_success_code_is_refused() {
    // Found while writing these tests: `Hresult(0)` would have injected
    // *success*, quietly falsifying this seam's central guarantee and letting
    // a test conceal the very defect it was written to find. The guarantee is
    // now enforced by a panic rather than asserted in prose.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let (_, completion) = real_completion(&mut ring);
    let _ = completion.with_injected_failure(crate::InjectedFailure::Hresult(0));
}

#[test]
fn every_named_condition_injects_a_genuine_failure() {
    // The other side of the guarantee: nothing a caller can spell through the
    // two *named* variants is capable of tripping the panic above, so the
    // enforcement constrains only the raw escape hatch.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let (_, completion) = real_completion(&mut ring);

    for condition in [
        crate::RingCondition::SubmissionQueueFull,
        crate::RingCondition::Corrupt,
        crate::RingCondition::VersionNotSupported,
    ] {
        assert!(
            completion
                .with_injected_failure(crate::InjectedFailure::Ring(condition))
                .result()
                .is_err(),
            "{condition:?} must inject a genuine failure"
        );
    }
    for code in [1_u32, 5, 87, 0xFFFF] {
        assert!(
            completion
                .with_injected_failure(crate::InjectedFailure::Win32(code))
                .result()
                .is_err(),
            "Win32({code}) must inject a genuine failure"
        );
    }
}

// --- opcode mapping, support probing, and the injection seam (M18.4) ---------

#[test]
fn every_op_maps_to_its_own_win32_opcode() {
    // `Op::code -> Default::default()` survived M18.3: nothing asserted the
    // mapping, only that operations built successfully.
    use windows_sys::Win32::Storage::FileSystem::{
        IORING_OP_CANCEL, IORING_OP_FLUSH, IORING_OP_NOP, IORING_OP_READ,
        IORING_OP_REGISTER_BUFFERS, IORING_OP_REGISTER_FILES, IORING_OP_WRITE,
    };

    assert_eq!(Op::Nop.code(), IORING_OP_NOP);
    assert_eq!(Op::Read.code(), IORING_OP_READ);
    assert_eq!(Op::Write.code(), IORING_OP_WRITE);
    assert_eq!(Op::Flush.code(), IORING_OP_FLUSH);
    assert_eq!(Op::Cancel.code(), IORING_OP_CANCEL);
    assert_eq!(Op::RegisterFiles.code(), IORING_OP_REGISTER_FILES);
    assert_eq!(Op::RegisterBuffers.code(), IORING_OP_REGISTER_BUFFERS);

    // Distinctness matters as much as the values: a mapping that collapsed two
    // operations onto one opcode would still satisfy each equality above if the
    // constants happened to agree.
    let mut codes: Vec<_> = Op::ALL.iter().map(|op| op.code()).collect();
    codes.sort_unstable();
    let before = codes.len();
    codes.dedup();
    assert_eq!(codes.len(), before, "every Op must have a distinct opcode");
}

#[test]
fn op_support_reads_the_bit_belonging_to_the_op_it_was_asked_about() {
    // `OpSupport::contains` finds the op's position in `Op::ALL` and tests that
    // bit. Replacing `==` with `!=` in the search survived M18.3, because every
    // op is supported on a healthy host and every answer was `true` regardless.
    // Constructed masks make the question observable.
    for (index, &op) in Op::ALL.iter().enumerate() {
        let only_this_one = OpSupport(1 << index);
        assert!(
            only_this_one.contains(op),
            "{op:?} must be reported supported when its own bit is set"
        );
        for other in Op::ALL {
            if other != op {
                assert!(
                    !only_this_one.contains(other),
                    "{other:?} must not be reported supported by {op:?}'s bit"
                );
            }
        }
    }

    let none = OpSupport(0);
    for op in Op::ALL {
        assert!(!none.contains(op));
    }
}

#[test]
fn a_reserved_opcode_is_not_supported() {
    // `IoRing::supports_raw -> true` survived because every opcode this crate
    // names is supported here. An opcode Win32 does not define is not.
    let ring = IoRing::new(8, 8).expect("create ring");
    assert!(!ring.supports_raw(0xFFFF));
    assert!(
        ring.supports_raw(Op::Read.code()),
        "the same call must still report a real opcode as supported"
    );
}

#[test]
fn an_injected_failure_carries_the_condition_it_names() {
    // The seam's own arithmetic: `Win32` wraps into an `HRESULT_FROM_WIN32`,
    // and a `Ring` condition keeps its documented code.
    use windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED;

    let ring = IoRing::new(8, 8).expect("create ring");
    let base = Completion::synthetic(9, 0, ring.ring_id());

    let win32 = base.with_injected_failure(InjectedFailure::Win32(ERROR_ACCESS_DENIED));
    let error = win32.result().expect_err("an injected failure must fail");
    // `check` wraps the HRESULT in a custom error rather than an OS one, so the
    // code comes back through the crate's own accessor.
    assert_eq!(
        error
            .as_ioring_error()
            .expect("an injected failure is an IoRing error")
            .code(),
        0x8007_0005_u32.cast_signed(),
        "Win32(ERROR_ACCESS_DENIED) must become HRESULT_FROM_WIN32 of that code"
    );

    let hresult = base.with_injected_failure(InjectedFailure::Hresult(-2_147_024_882));
    assert_eq!(
        hresult
            .result()
            .expect_err("an injected failure must fail")
            .as_ioring_error()
            .expect("an injected failure is an IoRing error")
            .code(),
        -2_147_024_882,
        "an HRESULT injection must be passed through unchanged"
    );

    // The transformation keeps the operation's identity, which is what makes
    // the seam sound: it rewrites a real completion rather than fabricating one.
    assert_eq!(win32.user_data(), base.user_data());
    assert_eq!(win32.ring_id(), base.ring_id());
}

#[test]
fn the_debug_rendering_names_the_ring_and_its_key_fields() {
    // `<impl Debug for IoRing>::fmt -> Ok(Default::default())` survived: that
    // mutation writes nothing to the formatter at all, so `format!("{ring:?}")`
    // comes back empty. Asserting the type name and a real field value is
    // enough to tell "wrote nothing" from "wrote the real struct".
    let ring = IoRing::new(8, 8).expect("create ring");
    let rendering = format!("{ring:?}");
    assert!(rendering.contains("IoRing"), "got {rendering}");
    assert!(
        rendering.contains("version"),
        "the version field name must appear: {rendering}"
    );
}

// ---------------------------------------------------------------------------
// M21.2: the bounded pop, and the wait it is generic over.
// ---------------------------------------------------------------------------

use super::{CompletionWait, RingWait, SubmitWait};

/// A scratch file to flush against, named per test so tests running as threads
/// in one process cannot collide on it.
fn pop_scratch(tag: &str) -> (std::path::PathBuf, std::fs::File) {
    let path = std::env::temp_dir().join(format!(
        "windows-ioring-sys-pop-within-{}-{tag}.tmp",
        std::process::id()
    ));
    std::fs::write(&path, b"x").expect("create fixture");
    let file = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(&path)
        .expect("open fixture");
    (path, file)
}

/// Push `count` flushes and submit them, popping nothing.
fn push_flushes(ring: &mut IoRing, file: &std::fs::File, count: usize) -> Vec<usize> {
    use crate::{Batch, FlushCoverage, FlushMode};
    use std::os::windows::io::AsRawHandle;

    let mut batch = Batch::new(ring);
    let mut ids = Vec::with_capacity(count);
    for _ in 0..count {
        // SAFETY: `file` outlives every operation pushed here -- the caller
        // drains before dropping it.
        ids.push(
            unsafe {
                batch.flush_raw(
                    file.as_raw_handle(),
                    FlushCoverage::Unordered,
                    FlushMode::Default,
                )
            }
            .expect("queue a flush"),
        );
    }
    batch.submit().expect("submit");
    ids
}

/// Records what the pop loop asked of it, and sleeps instead of blocking in
/// the ring.
///
/// Deliberately does **not** call [`RingWait::block`]. These tests drive the
/// loop with a reservation that has no real SQE behind it, and `SubmitIoRing`
/// answers `E_INVALIDARG` when asked to wait for a completion the kernel has
/// no pending operation for. Keeping the wait out of the kernel is what makes
/// the loop's own deadline behaviour testable without a slow real device.
#[derive(Default)]
struct RecordingWait {
    calls: usize,
    last_timeout_ms: u32,
    outstanding_seen: usize,
}

impl CompletionWait for RecordingWait {
    fn wait(&mut self, ring: &mut RingWait<'_>, timeout_ms: u32) -> std::io::Result<()> {
        self.calls += 1;
        self.last_timeout_ms = timeout_ms;
        self.outstanding_seen = ring.outstanding();
        std::thread::sleep(std::time::Duration::from_millis(2));
        Ok(())
    }
}

/// Refuses to wait at all, so a test can observe the error path.
struct FailingWait;

impl CompletionWait for FailingWait {
    fn wait(&mut self, _: &mut RingWait<'_>, _: u32) -> std::io::Result<()> {
        Err(std::io::Error::other("the wait refused"))
    }
}

/// Returns without blocking, which the trait explicitly permits.
struct ImmediateWait;

impl CompletionWait for ImmediateWait {
    fn wait(&mut self, _: &mut RingWait<'_>, _: u32) -> std::io::Result<()> {
        Ok(())
    }
}

#[test]
fn pop_within_returns_the_completion_of_a_real_operation() {
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let (path, file) = pop_scratch("real");
    let ids = push_flushes(&mut ring, &file, 1);

    let (completion, _held) = ring
        .pop_within(std::time::Duration::from_secs(30))
        .expect("pop_within")
        .expect("the flush completes well inside the bound");
    assert_eq!(
        completion.user_data(),
        ids[0],
        "the completion popped must be the one that was pushed"
    );
    let _ = std::fs::remove_file(&path);
}

#[test]
fn submit_wait_is_what_the_convenience_uses() {
    // The same operation through the explicit spelling. This is what proves
    // `SubmitWait` really does block in the ring: no other wait is involved,
    // and the completion still arrives.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let (path, file) = pop_scratch("submit-wait");
    let ids = push_flushes(&mut ring, &file, 1);

    let (completion, _held) = ring
        .pop_within_with(&mut SubmitWait, std::time::Duration::from_secs(30))
        .expect("pop_within_with")
        .expect("the flush completes");
    assert_eq!(completion.user_data(), ids[0]);
    let _ = std::fs::remove_file(&path);
}

#[test]
fn pop_within_returns_successive_completions_one_at_a_time() {
    let mut ring = IoRing::new(16, 32).expect("create ring");
    let (path, file) = pop_scratch("successive");
    let ids = push_flushes(&mut ring, &file, 3);

    let mut seen = Vec::new();
    for _ in 0..ids.len() {
        let (completion, _held) = ring
            .pop_within(std::time::Duration::from_secs(30))
            .expect("pop_within")
            .expect("each flush completes");
        seen.push(completion.user_data());
    }
    seen.sort_unstable();
    let mut expected = ids.clone();
    expected.sort_unstable();
    assert_eq!(
        seen, expected,
        "every pushed flush must be popped exactly once"
    );

    assert!(
        ring.pop_within(std::time::Duration::ZERO)
            .expect("pop_within")
            .is_none(),
        "the queue is empty once every completion has been taken"
    );
    let _ = std::fs::remove_file(&path);
}

#[test]
fn a_supplied_wait_is_not_consulted_when_nothing_can_arrive() {
    // The partner to `a_supplied_wait_is_consulted_...` below. A loop that
    // always waited once before checking would pass that test and fail this
    // one.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let mut wait = RecordingWait::default();
    let popped = ring
        .pop_within_with(&mut wait, std::time::Duration::from_secs(30))
        .expect("pop_within_with");
    assert!(popped.is_none());
    assert_eq!(
        wait.calls, 0,
        "nothing can arrive, so there is nothing to wait for"
    );
}

#[test]
fn a_supplied_wait_is_consulted_when_the_queue_is_not_ready() {
    let mut ring = IoRing::new(16, 16).expect("create ring");
    // A reservation with no real SQE behind it: outstanding, and no
    // completion will ever arrive for it.
    let op = ring.reserve_user_data().expect("reserve");
    let mut wait = RecordingWait::default();
    let popped = ring.pop_within_with(&mut wait, std::time::Duration::from_millis(40));
    // Settled before any assertion: a panic here would otherwise unwind into
    // `Drop`, whose rundown cannot settle a reservation the kernel never saw,
    // and the second panic would abort the whole harness.
    ring.record_completion(op);

    assert!(popped.expect("pop_within_with").is_none());
    assert!(
        wait.calls >= 1,
        "the supplied wait must be the thing that blocks"
    );
}

#[test]
fn the_deadline_is_honoured_when_an_operation_never_completes() {
    // No clock is consulted. That the loop *waited* rather than
    // short-circuiting is proved by the wait having been called; that it
    // *stopped* is proved by this test returning at all. A busy machine
    // changes how long that takes and changes neither assertion.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");

    let mut wait = RecordingWait::default();
    let popped = ring.pop_within_with(&mut wait, std::time::Duration::from_millis(40));
    ring.record_completion(op);

    assert!(
        popped.expect("pop_within_with").is_none(),
        "no completion was ever going to arrive"
    );
    assert!(
        wait.calls >= 1,
        "the bound must be waited out, not short-circuited"
    );
}

#[test]
fn a_zero_bound_does_not_block() {
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    let mut wait = RecordingWait::default();
    let popped = ring.pop_within_with(&mut wait, std::time::Duration::ZERO);
    ring.record_completion(op);

    assert!(popped.expect("pop_within_with").is_none());
    // The causal statement of "did not block": the wait is what blocks, and it
    // was never reached.
    assert_eq!(
        wait.calls, 0,
        "a zero bound is one try_pop, so the wait is never reached"
    );
}

#[test]
fn the_wait_is_never_handed_a_zero_timeout() {
    // A sub-millisecond remainder rounds to zero milliseconds, which
    // `SubmitIoRing` reads as "poll and return" -- turning the tail of every
    // bound into a spin. The loop clamps it up to one.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    let mut wait = RecordingWait::default();
    let popped = ring.pop_within_with(&mut wait, std::time::Duration::from_micros(600));
    ring.record_completion(op);

    assert!(popped.expect("pop_within_with").is_none());
    assert!(wait.calls >= 1, "the wait must have been reached at all");
    assert!(
        wait.last_timeout_ms >= 1,
        "a sub-millisecond remainder must be clamped up, never passed as zero"
    );
}

#[test]
fn ring_wait_reports_the_rings_outstanding_count() {
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let a = ring.reserve_user_data().expect("reserve a");
    let b = ring.reserve_user_data().expect("reserve b");
    let mut wait = RecordingWait::default();
    let popped = ring.pop_within_with(&mut wait, std::time::Duration::from_millis(20));
    ring.record_completion(a);
    ring.record_completion(b);

    assert!(popped.expect("pop_within_with").is_none());
    assert_eq!(
        wait.outstanding_seen, 2,
        "RingWait::outstanding must agree with IoRing::outstanding"
    );
}

#[test]
fn a_wait_that_fails_ends_the_pop_with_its_error() {
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    let outcome = ring.pop_within_with(&mut FailingWait, std::time::Duration::from_secs(30));
    ring.record_completion(op);

    let error = outcome.expect_err("the wait's failure must reach the caller");
    assert_eq!(error.to_string(), "the wait refused");
}

#[test]
fn a_wait_that_never_blocks_is_permitted_and_still_terminates() {
    // The trait says returning early is always allowed. A caller that does so
    // spins, which is their choice -- but the bound must still hold.
    //
    // **Termination is the assertion, and there is deliberately no clock.** An
    // earlier version asserted the elapsed time was under five seconds, which
    // could never have fired on the failure it named: if the deadline were not
    // honoured the loop would spin forever and that line would never be
    // reached. The only runs it could fail were slow ones -- so it was capable
    // of false failures and incapable of true ones. A loop that does not
    // terminate hangs the harness, which is what a hang looks like in every
    // other test here too.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    let popped = ring.pop_within_with(&mut ImmediateWait, std::time::Duration::from_millis(30));
    ring.record_completion(op);

    assert!(
        popped.expect("pop_within_with").is_none(),
        "the deadline is what ends a wait that never blocks"
    );
}

#[test]
fn a_bound_the_clock_cannot_represent_reaches_the_wait_rather_than_panicking() {
    // The half that exercises the overflow branch: something *is* outstanding,
    // so the loop computes a remaining duration from a deadline that could not
    // be represented. A failing wait is how the test escapes a bound that by
    // construction never arrives.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    let outcome = ring.pop_within_with(&mut FailingWait, std::time::Duration::MAX);
    ring.record_completion(op);

    let error = outcome.expect_err("the wait refuses, which is how this returns at all");
    assert_eq!(error.to_string(), "the wait refused");
}

#[test]
fn the_wait_can_be_supplied_as_a_trait_object() {
    // `?Sized` on the bound is what makes this compile, and a consumer
    // choosing a wait at run time is the reason to keep it.
    let mut ring = IoRing::new(16, 16).expect("create ring");
    let op = ring.reserve_user_data().expect("reserve");
    let wait: &mut dyn CompletionWait = &mut FailingWait;
    let outcome = ring.pop_within_with(wait, std::time::Duration::from_secs(30));
    ring.record_completion(op);

    let error = outcome.expect_err("a trait-object wait still refuses");
    assert_eq!(error.to_string(), "the wait refused");
}

/// A completion that retires no identity this ring is still owed is not
/// quiescence, and `run_down_within` must not report it as such -- for a raw
/// push exactly as for an owned one.
///
/// The first version of this guard lived inline in `Drop` and nowhere else, so
/// a saturating counter reaching zero still made `run_down_within` answer
/// `Ok(true)` -- "safe to drop". The second consulted the inventory as the
/// witness, which only owned pushes populate: with two RAW operations in
/// flight, a foreign CQE and the first real one drove the counter to zero over
/// an inventory that was empty by design, and rundown closed a ring the kernel
/// could still write through (PR #113 review). The ledger is now keyed by
/// identity, so neither case can reach `true`.
///
/// Reported as `Ok(false)` -- "not finished, call again" -- and NOT as an
/// error, because the operation still owed will complete (M10.2). Both
/// directions are pinned: once the real completions arrive, rundown does
/// report `true`, so a predicate that never quiesces would fail here too.
#[test]
fn a_completion_that_retires_nothing_is_not_quiescence() {
    use super::{Entry, Held};
    use crate::OperationId;
    use std::time::Duration;

    for owned in [false, true] {
        let mut ring = IoRing::refused_by_the_kernel();

        let first = ring.reserve_user_data().expect("mint the first");
        let second = ring.reserve_user_data().expect("mint the second");
        if owned {
            for user_data in [first, second] {
                ring.stow(
                    OperationId::new(user_data, ring.ring_id()),
                    Entry {
                        payload: None,
                        extra: (),
                        held: Held::default(),
                    },
                );
            }
        }

        // A completion for an identity this ring never minted, then the first
        // operation's real one, then a duplicate of it.
        assert!(
            !ring.record_completion(usize::MAX),
            "foreign (owned={owned})"
        );
        assert!(ring.record_completion(first), "real (owned={owned})");
        if owned {
            assert!(ring.reclaim(first).is_some(), "the first entry is retired");
        }
        assert!(!ring.record_completion(first), "duplicate (owned={owned})");

        assert_eq!(
            ring.outstanding(),
            1,
            "the second operation is still owed (owned={owned})"
        );
        let quiesced = ring
            .run_down_within(Duration::from_millis(0))
            .expect("a bound that expires is not an error");
        assert!(
            !quiesced,
            "rundown must not report quiescence while an operation is in flight (owned={owned})"
        );

        assert!(
            ring.record_completion(second),
            "the second's real completion"
        );
        if owned {
            assert!(
                ring.reclaim(second).is_some(),
                "the second entry is retired"
            );
        }
        let quiesced = ring
            .run_down_within(Duration::from_millis(0))
            .expect("nothing is outstanding, so nothing is submitted");
        assert!(
            quiesced,
            "with every identity retired, rundown must report quiescence (owned={owned})"
        );

        // Dropping this ring would assert on the null handle's close, which is
        // a sibling test's subject rather than this one's.
        std::mem::forget(ring);
    }
}

/// A kernel that answers one pop with a completion for `user_data`, then
/// reports the queue empty. Forwards nothing, so it is sound over the null
/// handle `IoRing::refused_by_the_kernel` builds.
#[cfg(feature = "kernel-seam")]
struct ForgedCompletion {
    user_data: Option<usize>,
}

#[cfg(feature = "kernel-seam")]
impl crate::sys::Responses for ForgedCompletion {
    unsafe fn pop(
        &mut self,
        _ring: *mut std::ffi::c_void,
        cqe: *mut windows_sys::Win32::Storage::FileSystem::IORING_CQE,
    ) -> windows_sys::core::HRESULT {
        let Some(user_data) = self.user_data.take() else {
            return windows_sys::Win32::Foundation::S_FALSE;
        };
        // SAFETY: the ring passes a valid out-pointer, as the Win32 call
        // requires.
        unsafe {
            cqe.write(windows_sys::Win32::Storage::FileSystem::IORING_CQE {
                UserData: user_data,
                ResultCode: 0,
                Information: 0,
            });
        }
        windows_sys::Win32::Foundation::S_OK
    }
}

/// A completion for an identity this ring never minted is a defect, and the
/// pop panics on it (D-79).
///
/// Driven through the real `pop_raw` by forging the kernel's answer, so what
/// is under test is the pop path every public pop shares, not the helper in
/// isolation. Covers the two causes the ring can see: an identity never minted
/// here, and a duplicate of one already retired.
#[cfg(feature = "kernel-seam")]
#[test]
#[should_panic(expected = "which is not in flight on this ring")]
fn a_completion_this_ring_never_minted_panics() {
    let mut ring = IoRing::refused_by_the_kernel();
    let _kernel = crate::sys::register(
        &ring,
        Box::new(ForgedCompletion {
            user_data: Some(0xDEAD),
        }),
    );
    let _ = ring.try_pop();
}

#[cfg(feature = "kernel-seam")]
#[test]
#[should_panic(expected = "which is not in flight on this ring")]
fn a_duplicate_of_a_retired_completion_panics() {
    let mut ring = IoRing::refused_by_the_kernel();
    let op = ring.reserve_user_data().expect("mint one identity");
    assert!(ring.record_completion(op), "its one real completion");

    let _kernel = crate::sys::register(
        &ring,
        Box::new(ForgedCompletion {
            user_data: Some(op),
        }),
    );
    let _ = ring.try_pop();
}

/// The other direction: a completion this ring *did* mint pops normally, so
/// the panic is about identity rather than about forged completions as such.
#[cfg(feature = "kernel-seam")]
#[test]
fn a_completion_this_ring_minted_does_not_panic() {
    let mut ring = IoRing::refused_by_the_kernel();
    let op = ring.reserve_user_data().expect("mint one identity");
    let kernel = crate::sys::register(
        &ring,
        Box::new(ForgedCompletion {
            user_data: Some(op),
        }),
    );

    // `pop_within` rather than `try_pop`: the forged kernel answers on the
    // first pop, but the crate's contract for "wait for this completion" is
    // the bounded pop, and the already-poppable census holds tests to it.
    let (completion, held) = ring
        .pop_within(std::time::Duration::from_secs(5))
        .expect("the forged pop succeeds")
        .expect("one completion was queued");
    assert_eq!(completion.user_data(), op);
    assert!(held.is_none(), "nothing was stowed for it");
    assert_eq!(ring.outstanding(), 0, "it retired the identity it carried");

    // Ended by hand: a forgotten ring never runs the drop that would end it.
    assert!(kernel.end().is_some());
    std::mem::forget(ring);
}

/// During an unwind the defect is traced but not raised: a second panic there
/// would abort the process, so the panic already in flight carries the
/// failure.
///
/// Rundown inside `Drop` is how this is reached for real. If the guard were
/// missing, this test would not fail -- the process would abort, which is the
/// behaviour the guard exists to exclude.
///
/// The `Drop` below records what it saw rather than asserting: any panic it
/// raised would itself be a second panic during the unwind, so a broken
/// precondition -- the seam not answering, say -- would abort the process and
/// hide which test failed. The assertions run after the unwind instead.
#[cfg(feature = "kernel-seam")]
#[test]
fn an_unminted_completion_during_an_unwind_does_not_panic_again() {
    type Seen = std::cell::RefCell<Option<(bool, Result<Option<usize>, String>)>>;

    struct PopsWhileUnwinding<'a> {
        ring: IoRing,
        seen: &'a Seen,
    }
    impl Drop for PopsWhileUnwinding<'_> {
        fn drop(&mut self) {
            let popped = self
                .ring
                .try_pop()
                .map(|popped| popped.map(|(completion, _)| completion.user_data()))
                .map_err(|error| error.to_string());
            *self.seen.borrow_mut() = Some((std::thread::panicking(), popped));
        }
    }

    let seen = Seen::default();
    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        let ring = IoRing::refused_by_the_kernel();
        let _kernel = crate::sys::register(
            &ring,
            Box::new(ForgedCompletion {
                user_data: Some(0xBEEF),
            }),
        );
        let _pops = PopsWhileUnwinding { ring, seen: &seen };
        panic!("the original failure");
    }));

    let payload = outcome.expect_err("the original panic propagates");
    assert_eq!(
        payload.downcast_ref::<&str>().copied(),
        Some("the original failure"),
        "the panic that surfaces is the original one, not the defect report"
    );
    let (unwinding, popped) = seen.take().expect("the drop ran");
    assert!(unwinding, "the pop happened during the unwind");
    assert_eq!(
        popped,
        Ok(Some(0xBEEF)),
        "the forged completion is still handed back"
    );
}