coordinode-lsm-tree 7.2.0

#![expect(
    clippy::expect_used,
    reason = "test assertions over known-good fixtures; failure surfaces via panic"
)]
#![expect(
    clippy::indexing_slicing,
    reason = "test code indexes fixture buffers with known sizes"
)]

use super::*;
use std::io::{Read, Write};
use std::sync::Arc;
// Shadows #[test] to enable log capture in test output.
use test_log::test;

/// Returns an `IoUringFs`, skipping only if the kernel lacks `io_uring`.
/// Constructor bugs (e.g. broken `RingThread::spawn`) will panic the
/// test instead of silently skipping.
fn try_io_uring() -> Option<IoUringFs> {
    if !is_io_uring_available() {
        eprintln!("skipping: io_uring not supported by kernel");
        return None;
    }
    // Kernel supports io_uring — constructor failures are real bugs.
    Some(IoUringFs::new().expect("io_uring available but IoUringFs::new() failed"))
}

#[test]
fn probe_availability() {
    // Just exercises the probe — result depends on the kernel.
    let available = is_io_uring_available();
    eprintln!("io_uring available: {available}");
}

#[test]
fn create_read_write() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let path = dir.path().join("test.txt");
    let opts = FsOpenOptions::new().write(true).create(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"hello world")?;
    file.sync_all()?;
    drop(file);

    let opts = FsOpenOptions::new().read(true);
    let mut file = fs.open(&path, &opts)?;
    let mut buf = String::new();
    file.read_to_string(&mut buf)?;
    assert_eq!(buf, "hello world");

    Ok(())
}

#[test]
fn read_at_pread_semantics() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let path = dir.path().join("pread.bin");
    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"hello world")?;
    file.sync_data()?;

    let mut buf = [0u8; 5];
    let n = file.read_at(&mut buf, 6)?;
    assert_eq!(n, 5);
    assert_eq!(&buf, b"world");

    let n = file.read_at(&mut buf, 0)?;
    assert_eq!(n, 5);
    assert_eq!(&buf, b"hello");

    Ok(())
}

#[test]
fn directory_operations() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let nested = dir.path().join("a").join("b").join("c");
    fs.create_dir_all(&nested)?;
    assert!(fs.exists(&nested)?);

    let file_path = nested.join("data.bin");
    let opts = FsOpenOptions::new().write(true).create_new(true);
    let mut file = fs.open(&file_path, &opts)?;
    file.write_all(b"data")?;
    drop(file);

    let entries: Vec<_> = fs.read_dir(&nested)?;
    assert_eq!(entries.len(), 1);
    assert_eq!(entries[0].file_name, "data.bin");

    let meta = fs.metadata(&file_path)?;
    assert!(meta.is_file);
    assert_eq!(meta.len, 4);

    fs.remove_file(&file_path)?;
    assert!(!fs.exists(&file_path)?);

    let top = dir.path().join("a");
    fs.remove_dir_all(&top)?;
    assert!(!fs.exists(&top)?);

    Ok(())
}

#[test]
fn rename() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let src = dir.path().join("src.txt");
    let dst = dir.path().join("dst.txt");

    let opts = FsOpenOptions::new().write(true).create(true);
    let mut file = fs.open(&src, &opts)?;
    file.write_all(b"content")?;
    drop(file);

    fs.rename(&src, &dst)?;
    assert!(!fs.exists(&src)?);
    assert!(fs.exists(&dst)?);

    Ok(())
}

#[test]
fn sync_directory() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    fs.sync_directory(dir.path())?;
    Ok(())
}

#[test]
fn file_metadata() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let path = dir.path().join("meta.bin");
    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"12345")?;

    let meta = file.metadata()?;
    assert!(meta.is_file);
    assert_eq!(meta.len, 5);

    Ok(())
}

#[test]
fn file_set_len() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let path = dir.path().join("truncate.bin");
    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"hello world")?;
    file.set_len(5)?;

    let meta = file.metadata()?;
    assert_eq!(meta.len, 5);

    Ok(())
}

#[test]
fn lock_exclusive() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;

    let path = dir.path().join("lockfile");
    let opts = FsOpenOptions::new().write(true).create(true);
    let file = fs.open(&path, &opts)?;
    file.lock_exclusive()?;

    Ok(())
}

#[test]
fn truncate_and_append() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("trunc.txt");

    let opts = FsOpenOptions::new().write(true).create(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"hello world")?;
    drop(file);

    let opts = FsOpenOptions::new().write(true).truncate(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"hi")?;
    drop(file);

    let meta = fs.metadata(&path)?;
    assert_eq!(meta.len, 2);

    let opts = FsOpenOptions::new().write(true).append(true);
    let mut file = fs.open(&path, &opts)?;
    // Seek to start, then write — append mode must ignore seek and
    // write at EOF regardless of cursor position.
    file.seek(SeekFrom::Start(0))?;
    file.write_all(b"!")?;
    drop(file);

    // Verify append went to EOF (len=3), not to start (which would
    // overwrite "hi" and keep len=2).
    let mut file = fs.open(&path, &FsOpenOptions::new().read(true))?;
    let mut buf = String::new();
    file.read_to_string(&mut buf)?;
    assert_eq!(buf, "hi!");
    assert_eq!(fs.metadata(&path)?.len, 3);

    Ok(())
}

#[test]
fn seek_operations() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("seek.bin");

    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"hello world")?;

    // Seek to start and re-read
    file.seek(SeekFrom::Start(0))?;
    let mut buf = [0u8; 5];
    file.read_exact(&mut buf)?;
    assert_eq!(&buf, b"hello");

    // Seek from current (+1 to skip space)
    file.seek(SeekFrom::Current(1))?;
    file.read_exact(&mut buf)?;
    assert_eq!(&buf, b"world");

    // Seek from end
    let pos = file.seek(SeekFrom::End(-5))?;
    assert_eq!(pos, 6);

    Ok(())
}

#[test]
fn concurrent_read_at() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("concurrent.bin");

    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    // Write 1000 bytes: each byte = (offset % 256)
    #[expect(clippy::cast_possible_truncation, reason = "% 256 guarantees 0..=255")]
    let data: Vec<u8> = (0u32..1000).map(|i| (i % 256) as u8).collect();
    file.write_all(&data)?;
    file.sync_all()?;

    let file = Arc::new(file);
    let mut handles = Vec::new();

    for chunk_start in (0..1000).step_by(100) {
        let file = Arc::clone(&file);
        handles.push(thread::spawn(move || -> io::Result<()> {
            let mut buf = [0u8; 100];
            let n = file.read_at(&mut buf, chunk_start as u64)?;
            assert_eq!(n, 100);
            for (i, &byte) in buf.iter().enumerate() {
                #[expect(clippy::cast_possible_truncation, reason = "% 256 guarantees 0..=255")]
                let expected = ((chunk_start + i) % 256) as u8;
                assert_eq!(byte, expected);
            }
            Ok(())
        }));
    }

    for h in handles {
        match h.join() {
            Ok(result) => result?,
            Err(_) => return Err(io::Error::other("thread panicked")),
        }
    }

    Ok(())
}

#[test]
fn metadata_directory() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let meta = fs.metadata(dir.path())?;
    assert!(meta.is_dir);
    assert!(!meta.is_file);

    Ok(())
}

#[test]
fn object_safety() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let fs: Arc<dyn Fs> = Arc::new(fs);
    let dir = tempfile::tempdir()?;
    let bogus = dir.path().join("nonexistent");
    assert!(!fs.exists(&bogus)?);
    Ok(())
}

#[test]
fn empty_buffer_returns_zero() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("empty_buf.bin");

    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"data")?;

    // read_at with empty buffer
    let n = file.read_at(&mut [], 0)?;
    assert_eq!(n, 0);

    // Read::read with empty buffer
    let n = file.read(&mut [])?;
    assert_eq!(n, 0);

    // Write::write with empty buffer
    let n = file.write(&[])?;
    assert_eq!(n, 0);

    // flush is a no-op
    file.flush()?;

    Ok(())
}

#[test]
fn sync_directory_rejects_file() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("not_a_dir.txt");

    let opts = FsOpenOptions::new().write(true).create(true);
    fs.open(&path, &opts)?;

    match fs.sync_directory(&path) {
        Ok(()) => panic!("sync_directory on a file should fail"),
        // sync_directory is an `Fs` method → returns `crate::io::Result`.
        Err(err) => assert_eq!(err.kind(), crate::io::ErrorKind::InvalidInput),
    }

    Ok(())
}

#[test]
fn seek_overflow_returns_error() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("seek_overflow.bin");

    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"data")?;

    // Seek to near u64::MAX, then seek forward — should overflow.
    file.seek(SeekFrom::Start(u64::MAX - 1))?;
    match file.seek(SeekFrom::Current(2)) {
        Ok(_) => panic!("seek past u64::MAX should fail"),
        Err(err) => assert_eq!(err.kind(), io::ErrorKind::InvalidInput),
    }

    // SeekFrom::Current negative past zero — should underflow.
    file.seek(SeekFrom::Start(0))?;
    match file.seek(SeekFrom::Current(-1)) {
        Ok(_) => panic!("seek before zero should fail"),
        Err(err) => assert_eq!(err.kind(), io::ErrorKind::InvalidInput),
    }

    // SeekFrom::End negative past zero — should underflow.
    match file.seek(SeekFrom::End(-100)) {
        Ok(_) => panic!("seek before zero should fail"),
        Err(err) => assert_eq!(err.kind(), io::ErrorKind::InvalidInput),
    }

    Ok(())
}

#[test]
fn debug_impl() {
    let Some(fs) = try_io_uring() else {
        return;
    };
    let debug = format!("{fs:?}");
    assert!(debug.contains("IoUringFs"));
}

#[test]
fn with_ring_size() -> io::Result<()> {
    if !is_io_uring_available() {
        eprintln!("skipping: io_uring not supported by kernel");
        return Ok(());
    }
    let fs =
        IoUringFs::with_ring_size(64).expect("io_uring available but with_ring_size(64) failed");
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("ring64.bin");
    let opts = FsOpenOptions::new().write(true).create(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"ok")?;
    file.sync_all()?;
    assert_eq!(fs.metadata(&path)?.len, 2);
    Ok(())
}

#[test]
fn seek_negative_from_current() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("seek_neg.bin");

    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut file = fs.open(&path, &opts)?;
    file.write_all(b"abcdefghij")?;

    // Seek to position 8, then back 3
    file.seek(SeekFrom::Start(8))?;
    let pos = file.seek(SeekFrom::Current(-3))?;
    assert_eq!(pos, 5);

    let mut buf = [0u8; 5];
    file.read_exact(&mut buf)?;
    assert_eq!(&buf, b"fghij");

    Ok(())
}

#[test]
fn clone_shares_ring() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let fs2 = fs.clone();
    let dir = tempfile::tempdir()?;

    // Both clones should work with the same ring thread.
    let p1 = dir.path().join("a.txt");
    let p2 = dir.path().join("b.txt");
    let opts = FsOpenOptions::new().write(true).create(true);

    let mut f1 = fs.open(&p1, &opts)?;
    let mut f2 = fs2.open(&p2, &opts)?;
    f1.write_all(b"one")?;
    f2.write_all(b"two")?;
    f1.sync_all()?;
    f2.sync_all()?;

    assert_eq!(fs.metadata(&p1)?.len, 3);
    assert_eq!(fs2.metadata(&p2)?.len, 3);

    Ok(())
}

#[test]
fn available_space_reports_plausible_free_bytes() -> io::Result<()> {
    // The cold-path free-space probe delegates to the shared statvfs helper:
    // the filesystem backing the tempdir must report a plausible, non-zero
    // figure below the unbounded sentinel.
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let free = fs.available_space(dir.path())?;
    assert!(
        free > 0,
        "a writable tempdir filesystem must report free space"
    );
    assert!(
        free < u64::MAX,
        "a real probe must not return the unbounded sentinel"
    );
    Ok(())
}

crate::fs::batch_conformance::batched_read_conformance!(super::try_io_uring());

#[test]
fn volume_id_matches_the_kernel_mount() -> io::Result<()> {
    // Free space is a property of the mount, not the I/O submission path, so
    // the uring backend reports the same volume id as `StdFs` for a path —
    // letting the space gate treat a uring data dir and a `StdFs` blob dir on
    // the same mount as one free-space pool.
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    assert_eq!(
        fs.volume_id(dir.path()),
        crate::fs::StdFs.volume_id(dir.path()),
        "uring and std agree on the mount backing a path"
    );
    assert!(fs.volume_id(dir.path()).is_some(), "a real mount has an id");
    Ok(())
}

/// Two files of `len` bytes each, byte `i` of file `f` being `(i + f) % 251`.
fn two_files(fs: &IoUringFs, dir: &Path, len: usize) -> io::Result<Vec<Box<dyn FsFile>>> {
    let opts = FsOpenOptions::new().write(true).create(true).read(true);
    let mut files = Vec::new();
    for f in 0..2usize {
        let mut file = fs.open(&dir.join(format!("f{f}.bin")), &opts)?;
        let bytes: Vec<u8> = (0..len)
            .map(|i| u8::try_from((i + f) % 251).expect("below 251"))
            .collect();
        file.write_all(&bytes)?;
        file.sync_all()?;
        files.push(file);
    }
    Ok(files)
}

/// What a caller spends per batch before the kernel sees it is one completion
/// channel and one acquisition of the submission lock, however many reads the
/// batch holds.
#[test]
fn a_batched_read_takes_one_channel_and_one_lock_per_batch() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files = two_files(&fs, dir.path(), 64 * 16)?;
    let mut buffers = vec![[0u8; 16]; 64];
    let mut reqs: Vec<_> = buffers
        .iter_mut()
        .enumerate()
        .map(|(i, buf)| crate::fs::BlockRead {
            file: files[i % 2].as_ref(),
            offset: (i * 16) as u64,
            buf: crate::fs::BlockBuf::new(buf),
        })
        .collect();

    let counts = &fs.inner.counts;
    let channels = counts.channels.load(core::sync::atomic::Ordering::Relaxed);
    let locks = counts.locks.load(core::sync::atomic::Ordering::Relaxed);
    fs.read_blocks_batched(&mut reqs)?;
    assert_eq!(
        counts.channels.load(core::sync::atomic::Ordering::Relaxed) - channels,
        1
    );
    assert_eq!(
        counts.locks.load(core::sync::atomic::Ordering::Relaxed) - locks,
        1
    );
    drop(reqs);
    for (i, buf) in buffers.iter().enumerate() {
        let expected: Vec<u8> = (0..16)
            .map(|b| u8::try_from((i * 16 + b + i % 2) % 251).expect("below 251"))
            .collect();
        assert_eq!(buf.as_slice(), expected, "block {i}");
    }
    Ok(())
}

/// `drain_batch` over completions fed in the given order: `(position, result)`.
fn drain_in(
    expected: &[usize],
    order: &[(usize, i32)],
) -> (Result<(), (usize, io::Error)>, Vec<usize>) {
    let mut feed = order.iter().copied();
    let mut handed = Vec::new();
    let verdict = drain_batch(expected, || feed.next(), "short", |p| handed.push(p));
    (verdict, handed)
}

/// Which failure a batch reports is a rule over the requests, not a race: the
/// lowest position, whatever order its completions arrive in.
#[test]
fn a_failed_batch_reports_the_lowest_failure_in_any_completion_order() {
    // Position 1 fails with EIO, position 3 reads short, position 4 fails with
    // ENOSPC; 0 and 2 succeed.
    let expected = [8, 8, 8, 8, 8];
    let events = [(0, 8), (1, -5), (2, 8), (3, 2), (4, -28)];
    let forward: Vec<_> = events.to_vec();
    let reverse: Vec<_> = events.iter().rev().copied().collect();
    let shuffled = vec![events[3], events[0], events[4], events[2], events[1]];

    for order in [forward, reverse, shuffled] {
        let (verdict, mut handed) = drain_in(&expected, &order);
        let Err((position, error)) = verdict else {
            panic!("a batch with failures must fail ({order:?})");
        };
        assert_eq!(position, 1, "{order:?}");
        assert_eq!(error.raw_os_error(), Some(5), "{order:?}");
        handed.sort_unstable();
        assert_eq!(
            handed,
            [0, 2],
            "only full reads are handed over ({order:?})"
        );
    }
}

/// A read that never completes, because the ring went away, is a broken pipe
/// at its position; a failure at a lower position still wins over it.
#[test]
fn a_batch_cut_short_reports_the_missing_read() {
    let (verdict, _) = drain_in(&[4, 4, 4], &[(0, 4), (2, 4)]);
    let Err((position, error)) = verdict else {
        panic!("a read that never completed must fail the batch");
    };
    assert_eq!(position, 1);
    assert_eq!(error.kind(), io::ErrorKind::BrokenPipe);

    let (verdict, _) = drain_in(&[4, 4, 4], &[(2, 4), (0, -5)]);
    assert!(matches!(verdict, Err((0, _))));
}

/// A finished read is handed over before later completions are even looked
/// at: the caller works on it while the rest are still in flight.
#[test]
fn a_finished_read_is_handed_over_before_later_completions_arrive() {
    let log = core::cell::RefCell::new(Vec::new());
    let mut feed = [(2, 4), (0, 4), (1, 4)].into_iter();
    let verdict = drain_batch(
        &[4, 4, 4],
        || {
            let next = feed.next();
            if let Some((position, _)) = next {
                log.borrow_mut().push(format!("complete {position}"));
            }
            next
        },
        "short",
        |position| log.borrow_mut().push(format!("hand {position}")),
    );
    assert!(verdict.is_ok());
    assert_eq!(
        log.into_inner(),
        [
            "complete 2",
            "hand 2",
            "complete 0",
            "hand 0",
            "complete 1",
            "hand 1"
        ]
    );
}

/// A callback that panics must not cut the drain short: the reads still in
/// flight write into buffers the unwind would free, so every completion is
/// received before the panic carries on.
#[test]
fn a_panicking_callback_still_drains_every_completion() {
    let received = core::cell::Cell::new(0usize);
    let mut feed = [(0, 4), (1, 4), (2, 4)].into_iter();
    let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        drain_batch(
            &[4, 4, 4],
            || {
                let next = feed.next();
                if next.is_some() {
                    received.set(received.get() + 1);
                }
                next
            },
            "short",
            |_| panic!("the caller's work on a finished read failed"),
        )
    }));
    assert!(unwound.is_err(), "the callback's panic carries on");
    assert_eq!(received.get(), 3, "every completion was received first");
}

/// The bytes a [`two_files`] file `f` holds at `offset`, `len` of them.
fn expected_bytes(f: usize, offset: usize, len: usize) -> Vec<u8> {
    (offset..offset + len)
        .map(|i| u8::try_from((i + f) % 251).expect("below 251"))
        .collect()
}

/// Reads submitted to the queue across two files come back with their tags
/// and bytes, and reads submitted after a wait, while earlier ones may still
/// be on the ring, are told apart from them.
#[test]
fn a_read_queue_hands_back_reads_of_several_submissions() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 64 * 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let mut queue = fs.read_queue();
    let mut got: Vec<(usize, Vec<u8>)> = Vec::new();
    for round in 0..4usize {
        for i in 0..16usize {
            let tag = round * 16 + i;
            queue.submit(QueuedRead {
                tag,
                file: Arc::clone(&files[tag % 2]),
                offset: (tag * 64) as u64,
                buf: vec![0; 64],
            });
        }
        // Issues this round's reads and takes whatever has come back.
        queue.wait(0, &mut |done| {
            assert!(done.result.is_ok(), "read {} failed", done.tag);
            got.push((done.tag, done.buf));
        });
    }
    while queue.outstanding() > 0 {
        queue.wait(1, &mut |done| {
            assert!(done.result.is_ok(), "read {} failed", done.tag);
            got.push((done.tag, done.buf));
        });
    }
    got.sort_unstable_by_key(|(tag, _)| *tag);
    let expected: Vec<(usize, Vec<u8>)> = (0..64)
        .map(|tag| (tag, expected_bytes(tag % 2, tag * 64, 64)))
        .collect();
    assert_eq!(got, expected);
    Ok(())
}

/// A read without a descriptor, which only a wait for some reads, is left
/// submitted by a wait for none, so a wrapper takes it back with the reads
/// that wait kept, and reads it at the wait that issues it.
#[test]
fn a_wait_for_none_keeps_a_read_without_a_descriptor_for_take_kept() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let mem = crate::fs::MemFs::new();
    mem.create_dir_all(Path::new("/d"))?;
    mem.open(
        Path::new("/d/f"),
        &FsOpenOptions::new().write(true).create(true),
    )?
    .write_all(&[7; 16])?;
    let file: Arc<dyn FsFile> =
        Arc::from(mem.open(Path::new("/d/f"), &FsOpenOptions::new().read(true))?);
    assert!(file.backing_fd().is_none(), "self-check: no descriptor");

    let mut queue = fs.read_queue();
    queue.submit(QueuedRead {
        tag: 3,
        file,
        offset: 0,
        buf: vec![0; 16],
    });
    let mut done = Vec::new();
    queue.wait(0, &mut |read| done.push(read));
    assert!(done.is_empty(), "a wait for none reads nothing serially");
    let kept = queue.take_kept();
    assert_eq!(
        kept.iter().map(|read| read.tag).collect::<Vec<_>>(),
        vec![3],
        "the read is handed back"
    );
    assert_eq!(queue.outstanding(), 0, "the queue no longer holds it");

    for read in kept {
        queue.submit(read);
    }
    queue.wait(1, &mut |read| done.push(read));
    assert_eq!(done.len(), 1);
    assert_eq!(done[0].buf, vec![7; 16]);
    Ok(())
}

/// Counts the wakes a queue gives it.
struct CountWake(core::sync::atomic::AtomicUsize);

impl crate::fs::ReadWake for CountWake {
    fn wake(&self) {
        self.0.fetch_add(1, core::sync::atomic::Ordering::SeqCst);
    }
}

/// A queue given a wake rings it once per read the ring finished, and only
/// once that read is ready: after `n` wakes, a wait that does not block hands
/// over at least `n` reads in all.
#[test]
fn a_read_queue_wakes_once_a_read_is_ready_to_hand_over() -> io::Result<()> {
    use core::sync::atomic::Ordering::SeqCst;

    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 32 * 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let wake = Arc::new(CountWake(core::sync::atomic::AtomicUsize::new(0)));
    let mut queue = fs.read_queue();
    assert!(queue.set_wake(Arc::clone(&wake) as Arc<dyn crate::fs::ReadWake>));
    for tag in 0..32usize {
        queue.submit(QueuedRead {
            tag,
            file: Arc::clone(&files[tag % 2]),
            offset: (tag * 64) as u64,
            buf: vec![0; 64],
        });
    }
    let mut handed = 0usize;
    while queue.outstanding() > 0 {
        let rung = wake.0.load(SeqCst);
        queue.wait(0, &mut |done| {
            assert!(done.result.is_ok(), "read {} failed", done.tag);
            handed += 1;
        });
        assert!(
            handed >= rung,
            "{rung} wakes rung but {handed} reads handed over"
        );
        std::thread::yield_now();
    }
    assert_eq!(handed, 32);
    assert_eq!(
        wake.0.load(SeqCst),
        32,
        "one wake per read the ring finished"
    );
    Ok(())
}

/// A read of a file the ring cannot take is read serially, which blocks, so
/// a wait for none leaves it held, and a wait for one reads it.
#[test]
fn a_read_without_a_descriptor_waits_for_a_wait_for_one() -> io::Result<()> {
    use crate::fs::MemFs;

    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let mem = MemFs::new();
    let path = Path::new("/m");
    mem.open(path, &FsOpenOptions::new().write(true).create(true))?
        .write_all(b"hello, world")?;
    let file: Arc<dyn FsFile> = Arc::from(mem.open(path, &FsOpenOptions::new().read(true))?);
    assert!(
        file.backing_fd().is_none(),
        "an in-memory file has no descriptor"
    );

    let mut queue = fs.read_queue();
    queue.submit(QueuedRead {
        tag: 1,
        file,
        offset: 7,
        buf: vec![0; 5],
    });
    let mut got = Vec::new();
    queue.wait(0, &mut |done| got.push(done.buf));
    assert!(got.is_empty(), "a wait for none reads nothing");
    assert_eq!((queue.outstanding(), queue.held()), (1, 1));
    queue.wait(1, &mut |done| got.push(done.buf));
    assert_eq!(got, [b"world".to_vec()]);
    assert_eq!((queue.outstanding(), queue.held()), (0, 0));
    Ok(())
}

/// A wait for none never blocks on the ring's submission channel: with the
/// ring thread held on a read that has no data yet and the channel full, it
/// keeps its reads and returns, and a later wait for one sends and reads them.
#[test]
fn a_wait_for_none_does_not_wait_for_room_on_the_ring() -> io::Result<()> {
    // Ring capacity, which bounds the submission channel.
    const ENTRIES: u32 = 2;
    let Some(_) = try_io_uring() else {
        return Ok(());
    };
    let fs = IoUringFs::with_ring_size(ENTRIES)?;
    let dir = tempfile::tempdir()?;
    let fifo = dir.path().join("fifo");
    let made = std::process::Command::new("mkfifo").arg(&fifo).status()?;
    assert!(made.success(), "mkfifo");
    // Read and write, so the open does not wait for a writer.
    let pipe: Arc<dyn FsFile> =
        Arc::from(fs.open(&fifo, &FsOpenOptions::new().read(true).write(true))?);
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let read_of = |file: &Arc<dyn FsFile>, len: usize| QueuedRead {
        tag: 0,
        file: Arc::clone(file),
        offset: 0,
        buf: vec![0; len],
    };

    // The ring thread takes this read and waits on it.
    let mut held = fs.read_queue();
    held.submit(read_of(&pipe, 1));
    held.wait(0, &mut |_| {});
    std::thread::sleep(std::time::Duration::from_millis(200));
    // Nothing drains the channel now: these fill it.
    let mut filling: Vec<_> = (0..ENTRIES)
        .map(|_| {
            let mut queue = fs.read_queue();
            queue.submit(read_of(&files[0], 64));
            queue.wait(0, &mut |_| {});
            queue
        })
        .collect();

    // The queue itself, which a thread can take, rather than the trait
    // object `read_queue` hands out.
    let mut late = UringReadQueue::new(&fs.inner);
    late.submit(read_of(&files[1], 64));
    std::thread::scope(|scope| {
        let (returned, polled) = std::sync::mpsc::channel();
        let poll = scope.spawn(move || {
            late.wait(0, &mut |_| {});
            returned.send(()).ok();
            late
        });
        let prompt = polled
            .recv_timeout(std::time::Duration::from_secs(5))
            .is_ok();
        // Release the ring thread either way, so the test ends.
        std::fs::OpenOptions::new()
            .write(true)
            .open(&fifo)
            .and_then(|mut writer| writer.write_all(b"x"))
            .expect("write to the fifo");
        let mut late = poll.join().expect("the poll thread");
        assert!(prompt, "a wait for none waited for room on the ring");

        assert_eq!(late.outstanding(), 1, "the read is kept, not lost");
        assert_eq!(
            late.held(),
            1,
            "a read no ring has is one only a wait carries out"
        );
        let mut got = Vec::new();
        while late.outstanding() > 0 {
            late.wait(1, &mut |done| got.push(done.result.is_ok()));
        }
        assert_eq!(got, [true], "a later wait sends and reads it");
    });
    for queue in &mut filling {
        while queue.outstanding() > 0 {
            queue.wait(1, &mut |done| assert!(done.result.is_ok()));
        }
    }
    while held.outstanding() > 0 {
        held.wait(1, &mut |done| assert!(done.result.is_ok()));
    }
    Ok(())
}

/// A wait for none does not wait for the submission lock either: while
/// another sender holds it (blocked on a full channel, say), the wait keeps
/// its reads, reports them held, and a later wait sends them.
#[test]
fn a_wait_for_none_does_not_wait_for_the_submission_lock() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let mut queue = UringReadQueue::new(&fs.inner);
    queue.submit(QueuedRead {
        tag: 0,
        file: Arc::clone(&files[0]),
        offset: 0,
        buf: vec![0; 64],
    });

    let lock = fs
        .inner
        .tx
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let mut queue = std::thread::scope(|scope| {
        let (returned, polled) = std::sync::mpsc::channel();
        let poll = scope.spawn(move || {
            queue.wait(0, &mut |_| {});
            returned.send(()).ok();
            queue
        });
        let prompt = polled
            .recv_timeout(std::time::Duration::from_secs(5))
            .is_ok();
        drop(lock);
        let queue = poll.join().expect("the poll thread");
        assert!(prompt, "a wait for none waited for the submission lock");
        queue
    });
    assert_eq!((queue.outstanding(), queue.held()), (1, 1));
    let mut got = Vec::new();
    while queue.outstanding() > 0 {
        queue.wait(1, &mut |done| got.push(done.result.is_ok()));
    }
    assert_eq!(got, [true], "a later wait sends and reads it");
    Ok(())
}

/// A wake asked for once reads are on the ring is declined: those reads
/// report to a sink without it, and a caller sleeping on it would not wake.
#[test]
fn a_wake_asked_for_with_reads_in_flight_is_declined() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let wake = Arc::new(CountWake(core::sync::atomic::AtomicUsize::new(0)));
    let mut queue = UringReadQueue::new(&fs.inner);
    queue.submit(QueuedRead {
        tag: 0,
        file: Arc::clone(&files[0]),
        offset: 0,
        buf: vec![0; 64],
    });
    queue.issue(true);
    assert!(!queue.set_wake(Arc::clone(&wake) as Arc<dyn crate::fs::ReadWake>));
    while queue.outstanding() > 0 {
        queue.wait(1, &mut |done| assert!(done.result.is_ok()));
    }
    assert!(
        queue.set_wake(wake as Arc<dyn crate::fs::ReadWake>),
        "nothing in flight"
    );
    Ok(())
}

/// A queue reused for many rounds keeps no slot for a read it handed back:
/// its bookkeeping follows the reads in flight, not every read it has sent.
#[test]
fn a_reused_read_queue_keeps_no_slot_for_a_read_handed_back() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 8 * 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let mut queue = UringReadQueue::new(&fs.inner);
    for round in 0..100usize {
        for i in 0..8usize {
            queue.submit(QueuedRead {
                tag: round * 8 + i,
                file: Arc::clone(&files[i % 2]),
                offset: (i * 64) as u64,
                buf: vec![0; 64],
            });
        }
        while queue.outstanding() > 0 {
            queue.wait(1, &mut |done| {
                assert!(done.result.is_ok(), "read {} failed", done.tag);
            });
        }
        assert!(
            queue.sent.is_empty(),
            "round {round} kept {} slots",
            queue.sent.len()
        );
    }
    Ok(())
}

/// The results of draining `queue` to empty, by tag, with each error's kind.
fn drain_results(queue: &mut dyn ReadQueue) -> Vec<(usize, Result<(), crate::io::ErrorKind>)> {
    let mut got = Vec::new();
    while queue.outstanding() > 0 {
        queue.wait(1, &mut |done| {
            got.push((done.tag, done.result.map_err(|e| e.kind())));
        });
    }
    got.sort_unstable_by_key(|(tag, _)| *tag);
    got
}

/// A read the ring need not take is handed back without it: an empty one at
/// once, and one too long for a submission queue entry as failed, beside a
/// read the ring serves.
#[test]
fn a_read_queue_hands_back_reads_the_ring_cannot_take() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 256)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let mut queue = fs.read_queue();
    queue.submit(QueuedRead {
        tag: 0,
        file: Arc::clone(&files[0]),
        offset: 0,
        buf: Vec::new(),
    });
    // Zeroed and never touched, so it costs address space, not memory.
    let too_long = usize::try_from(i32::MAX).expect("fits") + 1;
    queue.submit(QueuedRead {
        tag: 1,
        file: Arc::clone(&files[0]),
        offset: 0,
        buf: vec![0; too_long],
    });
    queue.submit(QueuedRead {
        tag: 2,
        file: Arc::clone(&files[0]),
        offset: 64,
        buf: vec![0; 64],
    });
    assert_eq!(
        drain_results(&mut *queue),
        [
            (0, Ok(())),
            (1, Err(crate::io::ErrorKind::InvalidInput)),
            (2, Ok(())),
        ]
    );
    Ok(())
}

/// A file with no descriptor is read serially, and what that read returns is
/// handed back as is: short of its block as failed, a refused read with its
/// error.
#[test]
fn a_serial_read_reports_a_short_read_and_a_refused_one() -> io::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, MemFs};

    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let mem = MemFs::new();
    let path = Path::new("/m");
    mem.open(path, &FsOpenOptions::new().write(true).create(true))?
        .write_all(b"hello, world")?;
    let short: Arc<dyn FsFile> = Arc::from(mem.open(path, &FsOpenOptions::new().read(true))?);
    let faulty = FaultFs::new(mem.clone());
    faulty.injector().arm(FaultRule::new(
        FaultOp::ReadAt,
        Fault::Error(crate::io::ErrorKind::PermissionDenied),
    ));
    let refused: Arc<dyn FsFile> = Arc::from(faulty.open(path, &FsOpenOptions::new().read(true))?);

    let mut queue = fs.read_queue();
    queue.submit(QueuedRead {
        tag: 0,
        file: short,
        offset: 7,
        buf: vec![0; 64],
    });
    queue.submit(QueuedRead {
        tag: 1,
        file: refused,
        offset: 0,
        buf: vec![0; 5],
    });
    assert_eq!(
        drain_results(&mut *queue),
        [
            (0, Err(crate::io::ErrorKind::UnexpectedEof)),
            (1, Err(crate::io::ErrorKind::PermissionDenied)),
        ]
    );
    Ok(())
}

/// Reads sent to a ring thread that has shut down fail, whether the wait
/// would block for room or not, and none is left outstanding.
#[test]
fn a_read_queue_fails_its_reads_once_the_ring_thread_is_gone() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 256)?
        .into_iter()
        .map(Arc::from)
        .collect();
    // What a shutdown leaves: no sender to the ring thread.
    drop(
        fs.inner
            .tx
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take(),
    );
    for min in [0usize, 1] {
        let mut queue = fs.read_queue();
        queue.submit(QueuedRead {
            tag: 0,
            file: Arc::clone(&files[0]),
            offset: 0,
            buf: vec![0; 64],
        });
        let mut got = Vec::new();
        queue.wait(min, &mut |done| {
            got.push((done.tag, done.result.map_err(|e| e.kind())));
        });
        assert_eq!(
            got,
            [(0, Err(crate::io::ErrorKind::BrokenPipe))],
            "a wait for {min}"
        );
        assert_eq!(queue.outstanding(), 0, "a wait for {min}");
    }
    Ok(())
}

/// A submission lock poisoned by a panicking sender still sends: the
/// channel it guards is intact, so a wait for none puts the reads on the ring.
#[test]
fn a_poisoned_submission_lock_still_sends() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 256)?
        .into_iter()
        .map(Arc::from)
        .collect();
    std::thread::scope(|scope| {
        let poisoner = scope.spawn(|| {
            let _held = fs
                .inner
                .tx
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            panic!("a sender panics holding the submission lock");
        });
        assert!(poisoner.join().is_err(), "the sender panicked");
    });
    assert!(fs.inner.tx.is_poisoned(), "the lock is poisoned");

    let mut queue = fs.read_queue();
    queue.submit(QueuedRead {
        tag: 0,
        file: Arc::clone(&files[0]),
        offset: 64,
        buf: vec![0; 64],
    });
    let mut got = Vec::new();
    while queue.outstanding() > 0 {
        queue.wait(0, &mut |done| got.push((done.tag, done.result.is_ok())));
        std::thread::yield_now();
    }
    assert_eq!(got, [(0, true)]);
    Ok(())
}

/// The ring's result for a read is handed back as the read's verdict: a
/// negative result is the read's `errno`. Completions may arrive in any order:
/// the slots of handed-back reads are dropped from either end, and a
/// completion for a read already handed back hands back nothing.
#[test]
fn a_read_queue_hands_back_completions_in_any_order() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 256)?
        .into_iter()
        .map(Arc::from)
        .collect();
    let mut queue = UringReadQueue::new(&fs.inner);
    // Two reads as if on the ring, at positions 0 and 1.
    for tag in 0..2usize {
        queue.sent.push_back(Some(QueuedRead {
            tag,
            file: Arc::clone(&files[0]),
            offset: 0,
            buf: vec![0; 64],
        }));
    }
    queue.on_ring = 2;

    /// `EACCES` on Linux.
    const EACCES: i32 = 13;
    // Every completion is taken before any assertion, so a failing one leaves
    // nothing for the queue's drop to wait for.
    let last = queue.complete(1, -EACCES);
    let slots_after_last = queue.sent.len();
    let again = queue.complete(1, 64);
    let first = queue.complete(0, 64);

    let last = last.expect("the read at 1");
    assert_eq!(
        last.result.map_err(|e| e.kind()),
        Err(crate::io::ErrorKind::PermissionDenied),
        "a negative result is the read's errno"
    );
    assert_eq!(
        slots_after_last, 1,
        "the handed-back slot at the end is dropped"
    );
    assert!(
        again.is_none(),
        "a completion for a read handed back hands back nothing"
    );
    assert!(first.expect("the read at 0").result.is_ok());
    assert!(queue.sent.is_empty(), "no slot outlives its read");
    assert_eq!(queue.on_ring, 0);
    Ok(())
}

/// Dropping a queue with reads in flight waits their completions out
/// before their buffers go, and the backend stays usable after.
#[test]
fn dropping_a_read_queue_with_reads_in_flight_leaves_the_backend_usable() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let files: Vec<Arc<dyn FsFile>> = two_files(&fs, dir.path(), 128 * 64)?
        .into_iter()
        .map(Arc::from)
        .collect();
    for _ in 0..50 {
        let mut queue = fs.read_queue();
        for tag in 0..128usize {
            queue.submit(QueuedRead {
                tag,
                file: Arc::clone(&files[tag % 2]),
                offset: (tag * 64) as u64,
                buf: vec![0; 64],
            });
        }
        // Puts every read on the ring without waiting for any.
        queue.wait(0, &mut |_| {});
        drop(queue);
    }
    let mut buf = [0u8; 64];
    let n = files[1].read_at(&mut buf, 64)?;
    assert_eq!(n, 64);
    assert_eq!(buf.to_vec(), expected_bytes(1, 64, 64));
    Ok(())
}

/// Writes `len` bytes of `byte` to a new file at `path` through `fs`.
fn write_file(fs: &IoUringFs, path: &Path, byte: u8, len: usize) -> io::Result<()> {
    let mut file = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
    file.write_all(&vec![byte; len])?;
    file.sync_all()?;
    Ok(())
}

/// A directory reports the same capabilities through `IoUringFs` as through
/// `StdFs`: both reach the same kernel files, and every path that adapts to
/// the file system (tight-space reclaim, prefix punches, reflinked
/// checkpoints) asks the capabilities first.
#[test]
fn capabilities_match_the_std_backend() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    assert_eq!(
        fs.capabilities(dir.path()),
        crate::fs::StdFs.capabilities(dir.path())
    );
    Ok(())
}

/// A hole punched through `IoUringFs` frees the range: it reads back as
/// zeros, the file keeps its length, and the range holds a hole.
#[test]
fn a_hole_punched_through_io_uring_lands() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    if !crate::fs::StdFs.capabilities(dir.path()).punch_hole {
        eprintln!("skipping: the temporary directory's file system cannot punch holes");
        return Ok(());
    }
    let path = dir.path().join("punched");
    write_file(&fs, &path, 0xAB, 1 << 20)?;
    fs.punch_hole(&path, 0, 512 * 1024)?;

    assert_eq!(fs.extent_contains_hole(&path, 0, 512 * 1024)?, Some(true));
    assert_eq!(fs.metadata(&path)?.len, 1 << 20, "the length is kept");
    let file = fs.open(&path, &FsOpenOptions::new().read(true))?;
    let mut head = [0xFFu8; 64];
    assert_eq!(file.read_at(&mut head, 0)?, 64);
    assert_eq!(head, [0u8; 64], "the punched range reads as zeros");
    let mut tail = [0u8; 64];
    assert_eq!(file.read_at(&mut tail, 512 * 1024)?, 64);
    assert_eq!(tail, [0xABu8; 64], "the rest is untouched");
    Ok(())
}

/// The link count of a path reads through `IoUringFs`, so a reclaim can tell
/// a file a checkpoint shares from one it may truncate.
#[test]
fn the_link_count_of_a_path_reads_through_io_uring() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let (path, link) = (dir.path().join("table"), dir.path().join("link"));
    write_file(&fs, &path, 1, 4_096)?;
    assert_eq!(Fs::hard_link_count(&fs, &path)?, 1);
    fs.hard_link(&path, &link)?;
    assert_eq!(Fs::hard_link_count(&fs, &path)?, 2);
    Ok(())
}

/// A file of `blocks` 64-byte blocks, byte `i` holding `i % 251`, opened for
/// reading through `fault` over the ring.
fn faulted_file(
    fs: &IoUringFs,
    fault: &crate::fs::FaultFs<IoUringFs>,
    path: &Path,
    blocks: usize,
) -> io::Result<Arc<dyn FsFile>> {
    let bytes: Vec<u8> = (0..blocks * 64)
        .map(|i| u8::try_from(i % 251).expect("below 251"))
        .collect();
    let mut file = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
    file.write_all(&bytes)?;
    file.sync_all()?;
    Ok(Arc::from(
        fault.open(path, &FsOpenOptions::new().read(true))?,
    ))
}

/// The bytes of block `block` of a [`faulted_file`].
fn faulted_block(block: usize) -> Vec<u8> {
    (block * 64..(block + 1) * 64)
        .map(|i| u8::try_from(i % 251).expect("below 251"))
        .collect()
}

/// Reads through a `FaultFs` over the ring go to the ring's queue together,
/// not one blocking read each, and a read rule still fires on the read it
/// names, the others unharmed.
#[test]
fn a_fault_rule_fires_on_reads_through_the_wrapped_ring_queue() -> io::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let fault = FaultFs::new(fs.clone());
    let dir = tempfile::tempdir()?;
    let file = faulted_file(&fs, &fault, &dir.path().join("queued"), 8)?;
    fault.injector().arm(
        FaultRule::new(FaultOp::ReadAt, Fault::Error(crate::io::ErrorKind::Other))
            .on_path("queued")
            .at_offset(3 * 64)
            .once(),
    );

    let channels = fs
        .inner
        .counts
        .channels
        .load(core::sync::atomic::Ordering::Relaxed);
    let mut done: Vec<(usize, Vec<u8>, Option<crate::io::ErrorKind>)> = Vec::new();
    {
        let mut queue = fault.read_queue();
        for tag in 0..8 {
            queue.submit(QueuedRead {
                tag,
                file: Arc::clone(&file),
                offset: (tag * 64) as u64,
                buf: vec![0; 64],
            });
        }
        queue.wait(8, &mut |read| {
            done.push((read.tag, read.buf, read.result.err().map(|e| e.kind())));
        });
    }
    assert_eq!(
        fs.inner
            .counts
            .channels
            .load(core::sync::atomic::Ordering::Relaxed)
            - channels,
        0,
        "no read took a blocking channel of its own"
    );
    done.sort_by_key(|(tag, ..)| *tag);
    assert_eq!(done.len(), 8);
    for (tag, buf, error) in done {
        if tag == 3 {
            assert_eq!(error, Some(crate::io::ErrorKind::Other), "the rule fired");
        } else {
            assert_eq!(error, None, "read {tag}");
            assert_eq!(buf, faulted_block(tag), "read {tag}");
        }
    }
    assert_eq!(fault.injector().read_count(), 8, "every read was seen");
    Ok(())
}

/// A batched read through a `FaultFs` over the ring is one batch of the ring.
#[test]
fn a_batched_read_through_a_fault_wrapper_is_one_batch_of_the_ring() -> io::Result<()> {
    use crate::fs::FaultFs;
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let fault = FaultFs::new(fs.clone());
    let dir = tempfile::tempdir()?;
    let file = faulted_file(&fs, &fault, &dir.path().join("batched"), 8)?;
    /// A read of each 64-byte block of `file` into its buffer.
    fn block_reads<'a>(
        file: &'a dyn FsFile,
        buffers: &'a mut [[u8; 64]],
    ) -> Vec<crate::fs::BlockRead<'a>> {
        buffers
            .iter_mut()
            .enumerate()
            .map(|(i, buf)| crate::fs::BlockRead {
                file,
                offset: (i * 64) as u64,
                buf: crate::fs::BlockBuf::new(buf),
            })
            .collect()
    }
    let mut buffers = vec![[0u8; 64]; 8];

    let channels = fs
        .inner
        .counts
        .channels
        .load(core::sync::atomic::Ordering::Relaxed);
    fault.read_blocks_batched(&mut block_reads(file.as_ref(), &mut buffers))?;
    assert_eq!(
        fs.inner
            .counts
            .channels
            .load(core::sync::atomic::Ordering::Relaxed)
            - channels,
        1,
        "one batch of the ring"
    );
    for (i, buf) in buffers.iter().enumerate() {
        assert_eq!(buf.to_vec(), faulted_block(i), "block {i}");
    }
    Ok(())
}

/// A file cloned through `IoUringFs` holds the source's bytes in an inode of
/// its own, not a second name of the source.
#[test]
fn a_file_cloned_through_io_uring_is_independent() -> io::Result<()> {
    let Some(fs) = try_io_uring() else {
        return Ok(());
    };
    let dir = tempfile::tempdir()?;
    let (source, clone) = (dir.path().join("source"), dir.path().join("clone"));
    write_file(&fs, &source, 7, 64 * 1024)?;
    fs.reflink_file(&source, &clone)?;

    let clone_file = fs.open(&clone, &FsOpenOptions::new().read(true).write(true))?;
    let mut bytes = vec![0u8; 64 * 1024];
    assert_eq!(clone_file.read_at(&mut bytes, 0)?, bytes.len());
    assert!(bytes.iter().all(|&b| b == 7), "the clone holds the source");
    assert_eq!(Fs::hard_link_count(&fs, &clone)?, 1, "an inode of its own");
    Ok(())
}