rustfs-uring 0.2.1

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Cancel-safety acceptance tests for the rustfs-uring read backend
//! (rustfs/backlog#894, #1048/#1051).
//!
//! In a restricted environment (Docker default seccomp, gVisor) the probe
//! fails with an expected-restriction errno and every test degrades to a
//! skip — the same contract P2's production probe must honor. Run under
//! `--security-opt seccomp=unconfined` (see run-docker.sh) to exercise the
//! real io_uring paths.
#![cfg(target_os = "linux")]

use std::fs::File;
use std::io::Write;
use std::os::fd::{AsRawFd, FromRawFd};
use std::sync::Arc;
use std::time::{Duration, Instant};

use rustfs_uring::UringDriver;

fn driver_or_skip(name: &str) -> Option<UringDriver> {
    match UringDriver::probe_and_start(64) {
        Ok(d) => Some(d),
        Err(e) => {
            assert!(
                e.is_expected_restriction(),
                "probe failed OUTSIDE the expected restriction errno class \
                 (EACCES/EPERM/ENOSYS/EINVAL/EOPNOTSUPP): {e:?}"
            );
            eprintln!("SKIP {name}: restricted environment, graceful degradation path taken ({e:?})");
            None
        }
    }
}

/// Like `driver_or_skip` but with `shards` independent rings (backlog#1145).
fn sharded_driver_or_skip(name: &str, shards: usize) -> Option<UringDriver> {
    match UringDriver::probe_and_start_sharded(64, shards) {
        Ok(d) => Some(d),
        Err(e) => {
            assert!(
                e.is_expected_restriction(),
                "probe failed OUTSIDE the expected restriction errno class: {e:?}"
            );
            eprintln!("SKIP {name}: restricted environment, graceful degradation path taken ({e:?})");
            None
        }
    }
}

/// Deterministic pseudo-random content so reads are verifiable.
fn make_content(len: usize) -> Vec<u8> {
    let mut state: u64 = 0x2545F4914F6CDD1D;
    (0..len)
        .map(|_| {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            state as u8
        })
        .collect()
}

fn temp_file_with(content: &[u8], tag: &str) -> (std::path::PathBuf, Arc<File>) {
    let path = std::env::temp_dir().join(format!("uring-spike-{tag}-{}", std::process::id()));
    std::fs::write(&path, content).expect("write temp file");
    let file = Arc::new(File::open(&path).expect("open temp file"));
    (path, file)
}

/// An OS pipe whose read side never completes until we write — the only
/// deterministic way to hold an op in flight across a future drop.
fn os_pipe() -> (Arc<File>, File) {
    let mut fds = [0i32; 2];
    // SAFETY: fds is a valid out-array; on success both fds are owned here
    // and immediately wrapped in File which takes over closing them.
    let rc = unsafe { libc::pipe(fds.as_mut_ptr()) };
    assert_eq!(rc, 0, "pipe(2) failed");
    let read = unsafe { File::from_raw_fd(fds[0]) };
    let write = unsafe { File::from_raw_fd(fds[1]) };
    (Arc::new(read), write)
}

async fn wait_until(deadline: Duration, mut cond: impl FnMut() -> bool) -> bool {
    let start = Instant::now();
    while start.elapsed() < deadline {
        if cond() {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(5)).await;
    }
    cond()
}

/// Baseline: completed reads return exactly what pread would.
#[tokio::test(flavor = "multi_thread")]
async fn read_matches_std() {
    let Some(driver) = driver_or_skip("read_matches_std") else {
        return;
    };
    const LEN: usize = 8 << 20;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "correctness");

    for i in 0..64usize {
        let offset = (i * 131_071) % (LEN - 70_000);
        let len = 1 + (i * 8_191) % 65_536;
        let got = driver
            .read_at(Arc::clone(&file), offset as u64, len)
            .await
            .expect("read failed");
        assert_eq!(got, &content[offset..offset + len], "mismatch at offset {offset} len {len}");
    }

    let snap = driver.shutdown();
    assert_eq!(snap.delivered, 64);
    assert_eq!(snap.orphan_reclaimed, 0);
    let _ = std::fs::remove_file(path);
}

/// THE core spike assertion: drop the future while the op is provably still
/// in flight (blocked pipe read, no cancel submitted) and verify the buffer
/// stays owned by the driver until the CQE finally arrives.
#[tokio::test(flavor = "multi_thread")]
async fn dropped_future_buffer_lives_until_cqe() {
    let Some(driver) = driver_or_skip("dropped_future_buffer_lives_until_cqe") else {
        return;
    };
    let (pipe_read, mut pipe_write) = os_pipe();

    let handle = driver.read_current(Arc::clone(&pipe_read), 4096).without_cancel_on_drop();
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await,
        "read never reached in-flight state"
    );

    drop(handle);

    // No cancel was submitted and the pipe is empty: the op MUST stay in
    // flight and the buffer MUST NOT be reclaimed, no matter how long the
    // future has been gone.
    tokio::time::sleep(Duration::from_millis(300)).await;
    let snap = driver.stats();
    assert_eq!(snap.in_flight, 1, "op vanished without a CQE");
    assert_eq!(snap.orphan_reclaimed, 0, "buffer reclaimed before CQE — UAF window!");

    // Now let the kernel complete the read; the CQE both writes into the
    // driver-owned buffer (safely) and triggers reclamation.
    pipe_write.write_all(&[0xAB; 512]).expect("pipe write");
    assert!(
        wait_until(Duration::from_secs(2), || {
            let s = driver.stats();
            s.in_flight == 0 && s.orphan_reclaimed == 1
        })
        .await,
        "orphaned op was not reclaimed at CQE: {:?}",
        driver.stats()
    );

    driver.shutdown();
}

/// Default drop path: ASYNC_CANCEL accelerates the CQE so the orphaned
/// buffer is reclaimed promptly without any data ever arriving.
#[tokio::test(flavor = "multi_thread")]
async fn async_cancel_accelerates_reclaim() {
    let Some(driver) = driver_or_skip("async_cancel_accelerates_reclaim") else {
        return;
    };
    let (pipe_read, pipe_write) = os_pipe();

    let handle = driver.read_current(Arc::clone(&pipe_read), 4096);
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await,
        "read never reached in-flight state"
    );

    drop(handle); // submits IORING_OP_ASYNC_CANCEL

    assert!(
        wait_until(Duration::from_secs(2), || {
            let s = driver.stats();
            s.in_flight == 0 && s.orphan_reclaimed == 1
        })
        .await,
        "cancel did not reclaim the orphan: {:?}",
        driver.stats()
    );

    drop(pipe_write);
    driver.shutdown();
}

/// Volume test modeling the EC quorum pattern: many concurrent reads, half
/// the futures dropped immediately. Every op must be accounted for as either
/// delivered or orphan-reclaimed, and survivors must return correct bytes.
#[tokio::test(flavor = "multi_thread")]
async fn cancel_stress_accounts_for_every_buffer() {
    let Some(driver) = driver_or_skip("cancel_stress_accounts_for_every_buffer") else {
        return;
    };
    const LEN: usize = 8 << 20;
    const OPS: usize = 256;
    const READ_LEN: usize = 64 << 10;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "stress");

    let mut kept = Vec::new();
    for i in 0..OPS {
        let offset = (i * 97_611) % (LEN - READ_LEN);
        let handle = driver.read_at(Arc::clone(&file), offset as u64, READ_LEN);
        if i % 2 == 0 {
            drop(handle); // dropped mid-flight or post-completion — both must be safe
        } else {
            kept.push((offset, handle));
        }
    }

    for (offset, handle) in kept {
        let got = handle.await.expect("kept read failed");
        assert_eq!(got, &content[offset..offset + READ_LEN], "mismatch at offset {offset}");
    }

    let snap = driver.shutdown();
    // `submitted == OPS` was an artifact of the old BLOCKING backpressure: the
    // 65th read_at parked the caller's thread until a permit freed, so every op
    // was submitted before its handle could be dropped. Under async backpressure
    // (rustfs/backlog#1102) a handle dropped before its first poll never
    // acquires a permit and therefore never submits — no buffer is allocated, so
    // there is nothing to reclaim (strictly better for memory). Ops that did get
    // an immediate permit are submitted eagerly and, when dropped, cancelled and
    // orphan-reclaimed.
    assert!(snap.submitted <= OPS as u64, "more ops submitted than requested: {snap:?}");
    // The exact split delivered == OPS/2 was flaky (C18, rustfs/backlog#1066):
    // an even-i read can finish between read_at returning and drop(handle),
    // delivering to the still-live receiver and flipping the split — all while
    // the ownership model behaves correctly. Only the conservation identity is
    // deterministic; the kept half guarantees delivered >= OPS/2.
    assert!(snap.delivered >= (OPS / 2) as u64, "kept half not all delivered: {snap:?}");
    // THE invariant: every op the driver accepted is accounted for — delivered
    // to a live caller or orphan-reclaimed at its CQE. No buffer is ever lost.
    assert_eq!(
        snap.delivered + snap.orphan_reclaimed,
        snap.submitted,
        "some buffers are unaccounted for: {snap:?}"
    );
    let _ = std::fs::remove_file(path);
}

/// Shutdown with ops still blocked in flight must cancel + drain them before
/// the driver thread exits (and the ring is unmapped).
#[tokio::test(flavor = "multi_thread")]
async fn shutdown_drains_in_flight_ops() {
    let Some(driver) = driver_or_skip("shutdown_drains_in_flight_ops") else {
        return;
    };
    let (pipe_read, pipe_write) = os_pipe();

    let h1 = driver.read_current(Arc::clone(&pipe_read), 1024);
    let h2 = driver.read_current(Arc::clone(&pipe_read), 1024);
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 2).await,
        "reads never reached in-flight state"
    );

    // shutdown() cancels both, drains to in_flight == 0 (asserted inside),
    // and joins the thread. The held futures then resolve with ECANCELED.
    let snap = driver.shutdown();
    assert_eq!(snap.in_flight, 0);
    assert_eq!(snap.delivered + snap.orphan_reclaimed, snap.submitted);

    for h in [h1, h2] {
        let err = h.await.expect_err("blocked pipe read cannot have succeeded");
        assert_eq!(err.raw_os_error(), Some(libc::ECANCELED), "unexpected error: {err:?}");
    }

    drop(pipe_write);
}

/// Invariant 2 regression (C20, rustfs/backlog#1064): the pending table — not
/// the caller — owns the fd. Deleting `Pending.file` used to leave every test
/// green because each test kept its own Arc<File> alive; this pins it. Drop the
/// caller's Arc while the op is in flight (bare drop, no cancel) and assert the
/// fd is STILL open: only the pending table's clone keeps it alive. If that
/// field were removed, the File would close the fd and F_GETFD returns EBADF.
#[tokio::test(flavor = "multi_thread")]
async fn pending_table_owns_fd_after_caller_drop() {
    let Some(driver) = driver_or_skip("pending_table_owns_fd_after_caller_drop") else {
        return;
    };
    let (pipe_read, mut pipe_write) = os_pipe();
    let raw_fd = pipe_read.as_raw_fd();

    let handle = driver.read_current(Arc::clone(&pipe_read), 64).without_cancel_on_drop();
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await,
        "read never reached in-flight state"
    );

    // The pending table is now the ONLY owner of the fd.
    drop(handle);
    drop(pipe_read);
    tokio::time::sleep(Duration::from_millis(50)).await;

    // SAFETY: F_GETFD only queries the descriptor; it neither closes nor
    // mutates it.
    let rc = unsafe { libc::fcntl(raw_fd, libc::F_GETFD) };
    assert_ne!(
        rc,
        -1,
        "fd was closed while an op is in flight — the pending table does not own it \
         (invariant 2 unprotected): {}",
        std::io::Error::last_os_error()
    );

    // Complete the op so the driver reclaims cleanly.
    pipe_write.write_all(&[0x5A; 64]).expect("pipe write");
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 0).await,
        "op was not reclaimed at CQE"
    );
    driver.shutdown();
}

/// Memory-safety integrity (C14, rustfs/backlog#1064): while an orphaned
/// blocked-pipe read holds a driver-owned buffer in flight the whole time, many
/// delivered reads must still come back byte-exact — the kernel writing into
/// the orphan's still-owned buffer must not corrupt anything, and the orphan
/// buffer must NOT be reclaimed before its own CQE. This is a direct
/// data-integrity observation, not just a counter identity. (A driver-level
/// poison/canary leg is a P2 acceptance-matrix item; ASAN cannot see a kernel
/// write into a freed buffer, so it is not the mechanism.)
#[tokio::test(flavor = "multi_thread")]
async fn orphan_in_flight_does_not_corrupt_delivered_reads() {
    let Some(driver) = driver_or_skip("orphan_in_flight_does_not_corrupt_delivered_reads") else {
        return;
    };
    let (pipe_read, mut pipe_write) = os_pipe();
    let orphan = driver.read_current(Arc::clone(&pipe_read), 4096).without_cancel_on_drop();
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await,
        "orphan never reached in-flight state"
    );
    drop(orphan); // bare drop: buffer stays owned by the driver until its CQE

    const LEN: usize = 1 << 20;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "canary");
    for i in 0..64usize {
        let offset = (i * 9_973) % (LEN - 4096);
        let got = driver
            .read_at(Arc::clone(&file), offset as u64, 4096)
            .await
            .expect("read failed");
        assert_eq!(got, &content[offset..offset + 4096], "delivered read corrupted at offset {offset}");
    }
    assert_eq!(driver.stats().orphan_reclaimed, 0, "orphan buffer reclaimed before its CQE");

    pipe_write.write_all(&[0u8; 512]).expect("pipe write");
    driver.shutdown();
    let _ = std::fs::remove_file(path);
}

/// CQ-overflow safety (C15, rustfs/backlog#1065). With backpressure capping
/// in-flight at the SQ depth (64) below CQ capacity (128), overflow is
/// structurally unreachable. Drive far more ops than CQ capacity through and
/// assert the kernel overflow counter stays 0 and every op is delivered.
#[tokio::test(flavor = "multi_thread")]
async fn no_cq_overflow_under_load() {
    let Some(driver) = driver_or_skip("no_cq_overflow_under_load") else {
        return;
    };
    const LEN: usize = 8 << 20;
    const OPS: usize = 300; // > CQ capacity (128) many times over
    const READ_LEN: usize = 4096;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "overflow");

    let mut kept = Vec::new();
    for i in 0..OPS {
        let offset = (i * 4_093) % (LEN - READ_LEN);
        kept.push((offset, driver.read_at(Arc::clone(&file), offset as u64, READ_LEN)));
    }
    for (offset, handle) in kept {
        let got = handle.await.expect("read failed");
        assert_eq!(got, &content[offset..offset + READ_LEN], "mismatch at offset {offset}");
    }

    let snap = driver.shutdown();
    assert_eq!(snap.cq_overflow, 0, "CQ overflowed under load: {snap:?}");
    assert_eq!(snap.delivered, OPS as u64);
    let _ = std::fs::remove_file(path);
}

/// Boundary reads on a regular file (C16, rustfs/backlog#1065): len==0, read at
/// EOF, a cross-EOF short read delivered to a live receiver (exercises the C9
/// resubmit loop), and the rejected huge-len / huge-offset guards (C6/C7).
#[tokio::test(flavor = "multi_thread")]
async fn boundary_reads() {
    let Some(driver) = driver_or_skip("boundary_reads") else {
        return;
    };
    const LEN: usize = 4096;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "boundary");

    // len == 0 → empty Ok.
    let got = driver.read_at(Arc::clone(&file), 0, 0).await.expect("len=0 read");
    assert!(got.is_empty(), "len=0 should return empty, got {}", got.len());

    // offset == file size → EOF → empty Ok.
    let got = driver.read_at(Arc::clone(&file), LEN as u64, 128).await.expect("EOF read");
    assert!(got.is_empty(), "read at EOF should be empty, got {}", got.len());

    // Read spanning past EOF → the available tail bytes, delivered to a live
    // receiver via the resubmit-then-EOF path.
    let got = driver
        .read_at(Arc::clone(&file), (LEN - 10) as u64, 100)
        .await
        .expect("cross-EOF read");
    assert_eq!(got, &content[LEN - 10..LEN], "cross-EOF read should return the tail only");

    // len > MAX_RW_COUNT → rejected (C6); offset > i64::MAX → rejected (C7).
    let err = driver
        .read_at(Arc::clone(&file), 0, (1usize << 32) + 1)
        .await
        .expect_err("huge len must be rejected");
    assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput, "huge len error: {err:?}");
    let err = driver
        .read_at(Arc::clone(&file), (i64::MAX as u64) + 1, 16)
        .await
        .expect_err("huge offset must be rejected");
    assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput, "huge offset error: {err:?}");

    driver.shutdown();
    let _ = std::fs::remove_file(path);
}

/// Pipe half-close boundary (C16, rustfs/backlog#1065): an in-flight read whose
/// write end is closed observes EOF (res=0) and returns empty.
#[tokio::test(flavor = "multi_thread")]
async fn pipe_half_close_reads_eof() {
    let Some(driver) = driver_or_skip("pipe_half_close_reads_eof") else {
        return;
    };
    let (pipe_read, pipe_write) = os_pipe();
    let handle = driver.read_current(Arc::clone(&pipe_read), 128);
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 1).await,
        "read never reached in-flight state"
    );
    drop(pipe_write); // close write end → blocked read observes EOF
    let got = handle.await.expect("pipe EOF read");
    assert!(got.is_empty(), "closed-pipe read should be empty EOF, got {}", got.len());
    driver.shutdown();
}

/// Open `path` read-only with `O_DIRECT`. `None` ONLY when the filesystem
/// genuinely refuses O_DIRECT (tmpfs, some overlay/network mounts return EINVAL
/// or EOPNOTSUPP), which is a legitimate configuration to skip on. Any other
/// error (ENOENT, EMFILE, EACCES, …) is a real test bug and panics loudly so it
/// can't masquerade as a skipped assertion; EINTR is retried.
fn open_direct(path: &std::path::Path) -> Option<Arc<File>> {
    use std::os::unix::ffi::OsStrExt;
    let c_path = std::ffi::CString::new(path.as_os_str().as_bytes()).expect("temp path has no NUL");
    loop {
        // SAFETY: `c_path` is a valid NUL-terminated path; on success we own the fd.
        let fd = unsafe { libc::open(c_path.as_ptr(), libc::O_RDONLY | libc::O_DIRECT | libc::O_CLOEXEC) };
        if fd >= 0 {
            return Some(Arc::new(unsafe { File::from_raw_fd(fd) }));
        }
        let err = std::io::Error::last_os_error();
        match err.raw_os_error() {
            Some(libc::EINTR) => continue,
            Some(libc::EINVAL) | Some(libc::EOPNOTSUPP) => return None,
            _ => panic!("open({}, O_DIRECT) failed unexpectedly: {err}", path.display()),
        }
    }
}

/// O_DIRECT positioned read (rustfs/backlog#1102): the driver reads the
/// block-aligned superset range into a block-aligned buffer and hands back
/// exactly the requested logical range. Alignment padding, the bytes before the
/// range, and the block-aligned tail must never escape — a BitrotReader
/// expecting an exact shard length would flag padded output as corruption.
#[tokio::test(flavor = "multi_thread")]
async fn direct_read_returns_exact_unaligned_ranges() {
    let Some(driver) = driver_or_skip("direct_read_returns_exact_unaligned_ranges") else {
        return;
    };

    const ALIGN: usize = 4096;
    const FILE_LEN: usize = ALIGN * 3 + 7; // deliberately not block-aligned
    let content = make_content(FILE_LEN);

    let path = std::env::temp_dir().join(format!("uring-odirect-{}", std::process::id()));
    {
        let mut f = File::create(&path).expect("create temp file");
        f.write_all(&content).expect("write temp file");
        // O_DIRECT bypasses the page cache; make sure the data is on the device.
        f.sync_all().expect("sync temp file");
    }

    let Some(file) = open_direct(&path) else {
        // Not a skip of the io_uring paths — the suite still exercised them —
        // only the O_DIRECT assertions cannot run on this filesystem.
        eprintln!("direct_read_returns_exact_unaligned_ranges: filesystem rejects O_DIRECT, assertions not exercised");
        driver.shutdown();
        let _ = std::fs::remove_file(&path);
        return;
    };

    // Unaligned starts, block-straddling ranges, and the unaligned file tail.
    let ranges = [
        (0usize, FILE_LEN),
        (0, ALIGN),
        (1, 17),
        (ALIGN - 1, ALIGN + 2),
        (ALIGN * 2, ALIGN + 7),
        (FILE_LEN - 7, 7),
    ];
    for (offset, len) in ranges {
        let got = driver
            .read_at_direct(Arc::clone(&file), offset as u64, len, ALIGN)
            .await
            .expect("O_DIRECT read failed");
        assert_eq!(got, &content[offset..offset + len], "O_DIRECT read mismatch at offset={offset} len={len}");
    }

    // A non-power-of-two alignment is rejected before the kernel sees it.
    let err = driver
        .read_at_direct(Arc::clone(&file), 0, 16, 3)
        .await
        .expect_err("non-power-of-two alignment must be rejected");
    assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput, "unexpected error: {err:?}");

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

/// Async backpressure (rustfs/backlog#1102): once every permit is held, `submit`
/// must NOT park the caller's thread. It returns a handle that has not submitted
/// anything; the permit is acquired — and the op submitted — on the handle's
/// first poll, after a permit frees.
///
/// With the old blocking backpressure this test would hang: the saturating reads
/// are blocked pipe reads, so no permit can ever free while `read_at` parks.
#[tokio::test(flavor = "multi_thread")]
async fn saturated_submit_defers_instead_of_blocking() {
    const DEPTH: u32 = 4;
    let driver = match UringDriver::probe_and_start(DEPTH) {
        Ok(d) => d,
        Err(e) => {
            assert!(
                e.is_expected_restriction(),
                "probe failed OUTSIDE the expected restriction errno class: {e:?}"
            );
            eprintln!("SKIP saturated_submit_defers_instead_of_blocking: restricted environment ({e:?})");
            return;
        }
    };

    let (pipe_read, mut pipe_write) = os_pipe();
    const LEN: usize = 4096;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "backpressure");

    // Saturate: DEPTH blocked pipe reads hold every permit.
    let mut held = Vec::new();
    for _ in 0..DEPTH {
        held.push(driver.read_current(Arc::clone(&pipe_read), 64));
    }
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == u64::from(DEPTH)).await,
        "backpressure never saturated: {:?}",
        driver.stats()
    );

    // Saturated. This call must return immediately (it would block forever under
    // the old implementation) and must NOT have submitted anything.
    let extra = driver.read_at(Arc::clone(&file), 0, LEN);
    let snap = driver.stats();
    assert_eq!(snap.in_flight, u64::from(DEPTH), "saturated submit put an op in flight: {snap:?}");
    assert_eq!(snap.submitted, u64::from(DEPTH), "saturated submit submitted an op: {snap:?}");

    // Free the permits by completing the blocked pipe reads.
    pipe_write
        .write_all(&vec![0xAB; 64 * (DEPTH as usize + 1)])
        .expect("pipe write");
    for h in held {
        let _ = h.await;
    }

    // The deferred handle acquires a freed permit on its first poll, submits,
    // and returns the correct bytes.
    let got = extra.await.expect("deferred read must complete once a permit frees");
    assert_eq!(got, content, "deferred read returned wrong bytes");
    assert_eq!(driver.stats().submitted, u64::from(DEPTH) + 1);

    driver.shutdown();
    let _ = std::fs::remove_file(path);
}

/// Drop-without-shutdown path (C17, rustfs/backlog#1066): every other test ends
/// via explicit shutdown(), so the UringDriver `Drop` impl's live-thread branch
/// was never exercised. Drop the driver directly with ops still in flight; the
/// Drop must send Shutdown BEFORE joining (send-after-join would deadlock at
/// recv), cancel + drain the in-flight ops, and join — the held futures then
/// resolve to ECANCELED. A regression (join-first, or an unbounded hang) makes
/// this test hang.
#[tokio::test(flavor = "multi_thread")]
async fn drop_without_shutdown_drains_and_cancels() {
    let Some(driver) = driver_or_skip("drop_without_shutdown_drains_and_cancels") else {
        return;
    };
    let (pipe_read, pipe_write) = os_pipe();
    let h1 = driver.read_current(Arc::clone(&pipe_read), 1024);
    let h2 = driver.read_current(Arc::clone(&pipe_read), 1024);
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == 2).await,
        "reads never reached in-flight state"
    );

    // No shutdown() — exercise Drop directly.
    drop(driver);

    for h in [h1, h2] {
        let err = h.await.expect_err("blocked pipe read cannot have succeeded");
        assert_eq!(err.raw_os_error(), Some(libc::ECANCELED), "unexpected error: {err:?}");
    }
    drop(pipe_write);
}

/// Sharding the driver's rings (backlog#1145) must not weaken the cancel-safety
/// model: every op the driver accepted is still delivered to a live caller or
/// orphan-reclaimed at its CQE, and the aggregate counters conserve. Ops are
/// round-robined across shards, so a cancel routed to the wrong ring — or a
/// pending table that lost an entry across the split — shows up here as a
/// conservation violation or a content mismatch.
///
/// Deliberately mixes dropped and kept handles, exactly as `cancel_stress` does,
/// but across 4 rings so the drop lands on a shard other than the one the next
/// submit picks.
#[tokio::test(flavor = "multi_thread")]
async fn sharded_driver_conserves_buffers_across_shards() {
    const SHARDS: usize = 4;
    let Some(driver) = sharded_driver_or_skip("sharded_driver_conserves_buffers_across_shards", SHARDS) else {
        return;
    };
    const LEN: usize = 8 << 20;
    const OPS: usize = 256;
    const READ_LEN: usize = 64 << 10;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "sharded-stress");

    let mut kept = Vec::new();
    for i in 0..OPS {
        let offset = (i * 97_611) % (LEN - READ_LEN);
        let handle = driver.read_at(Arc::clone(&file), offset as u64, READ_LEN);
        if i % 2 == 0 {
            drop(handle);
        } else {
            kept.push((offset, handle));
        }
    }

    // Correctness across shards: a kept read must return its own bytes, not a
    // neighbouring shard's.
    for (offset, handle) in kept {
        let got = handle.await.expect("kept read failed");
        assert_eq!(got, &content[offset..offset + READ_LEN], "mismatch at offset {offset}");
    }

    let snap = driver.shutdown();
    assert!(snap.submitted <= OPS as u64, "more ops submitted than requested: {snap:?}");
    assert!(snap.delivered >= (OPS / 2) as u64, "kept half not all delivered: {snap:?}");
    assert_eq!(
        snap.delivered + snap.orphan_reclaimed,
        snap.submitted,
        "some buffers are unaccounted for across shards: {snap:?}"
    );
    assert_eq!(snap.in_flight, 0, "shutdown left ops in flight: {snap:?}");
    assert_eq!(snap.cq_overflow, 0, "per-shard in-flight cap must keep CQ overflow unreachable: {snap:?}");
    let _ = std::fs::remove_file(path);
}

/// A single-shard sharded driver must behave exactly like `probe_and_start`,
/// which is what every other test in this file exercises.
#[tokio::test(flavor = "multi_thread")]
async fn sharded_driver_with_one_shard_matches_single_ring() {
    let Some(driver) = sharded_driver_or_skip("sharded_driver_with_one_shard_matches_single_ring", 1) else {
        return;
    };
    const LEN: usize = 1 << 20;
    let content = make_content(LEN);
    let (path, file) = temp_file_with(&content, "sharded-one");
    let got = driver.read_at(Arc::clone(&file), 4096, 8192).await.expect("read");
    assert_eq!(got, &content[4096..4096 + 8192]);
    let snap = driver.shutdown();
    assert_eq!(snap.delivered + snap.orphan_reclaimed, snap.submitted, "{snap:?}");
    let _ = std::fs::remove_file(path);
}

/// Deterministic sharded cancel routing (backlog#1180). A `ReadHandle` carries
/// the tx/wake of the shard that accepted its op, so a drop-cancel must reach
/// THAT shard's ring. The existing sharded stress test cannot prove this: its
/// dropped regular-file reads complete on their own, so a mis-routed cancel
/// (answered -ENOENT) still conserves. Here every op is a blocked-pipe read that
/// can ONLY finish by being canceled, and there is more than one op per shard, so
/// a cancel that landed on the wrong ring would leave its read stuck and
/// in_flight would never reach 0.
#[tokio::test(flavor = "multi_thread")]
async fn sharded_cancel_routes_to_the_owning_ring() {
    const SHARDS: usize = 4;
    // Comfortably more than SHARDS so round-robin gives every shard at least one.
    const OPS: usize = 16;
    let Some(driver) = sharded_driver_or_skip("sharded_cancel_routes_to_the_owning_ring", SHARDS) else {
        return;
    };
    let (pipe_read, pipe_write) = os_pipe();

    let mut handles = Vec::new();
    for _ in 0..OPS {
        handles.push(driver.read_current(Arc::clone(&pipe_read), 4096));
    }
    assert!(
        wait_until(Duration::from_secs(2), || driver.stats().in_flight == OPS as u64).await,
        "not all reads reached in-flight: {:?}",
        driver.stats()
    );

    // Drop every handle: each sends AsyncCancel to the shard that accepted it.
    drop(handles);

    assert!(
        wait_until(Duration::from_secs(3), || {
            let s = driver.stats();
            s.in_flight == 0 && s.orphan_reclaimed == OPS as u64
        })
        .await,
        "sharded cancels did not reclaim every orphan — a cancel routed to the wrong ring? {:?}",
        driver.stats()
    );
    // Blocked-pipe reads cannot complete on their own, so reclaiming every one
    // proves the cancels — not a timeout — did it.
    let snap = driver.stats();
    assert!(snap.cancel_succeeded > 0, "orphans not reclaimed by cancel: {snap:?}");

    drop(pipe_write);
    driver.shutdown();
}