krabka_log/

io.rs

//! Injectable file I/O for every durable write the log makes.
//!
//! The active `.log` file keeps the three handle-shaped methods the append
//! hot path uses. Every other durable write in the crate -- the sparse
//! indexes, the `.stampindex`, the producer snapshots, the leader-epoch
//! checkpoint, the compaction swap and the segment deletions retention
//! performs -- goes through the path-and-target methods below, so a test can
//! fail exactly one class of file and watch what recovery makes of it. The
//! readahead hint of the fetch path goes through it too, so a test can see
//! which ranges a read asks the kernel for, and so do the block reservation
//! and release of `preallocate` and the `O_DIRECT` handle its active segment
//! writes through.

#[cfg(any(target_os = "linux", test))]
use std::iter;
#[cfg(unix)]
use std::os::unix::fs::FileExt;
use std::{
    fmt::Debug,
    fs::{self, File},
    io::{self, ErrorKind, IoSlice, Write},
    path::Path,
    sync::Arc,
};
#[cfg(target_os = "linux")]
use {
    rustix::{
        fs::{
            Advice, FallocateFlags, SyncFileRangeFlags, fadvise, fallocate, open, statx,
            sync_file_range,
        },
        io::{Errno, ReadWriteFlags, preadv2},
    },
    std::num::NonZeroU64,
};

/// Which on-disk file class an [`LogIo`] operation touches.
///
/// A fault injector matches on this to fail one durable write without
/// disturbing the rest of the log.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IoTarget {
    /// A segment's sparse `.index`.
    OffsetIndex,
    /// A segment's sparse `.timeindex`.
    TimeIndex,
    /// A segment's `.stampindex` sidecar.
    StampIndex,
    /// A `<offset>.snapshot` producer-state snapshot, or its `.tmp` staging
    /// file.
    ProducerSnapshot,
    /// The partition's `leader-epoch-checkpoint`, or its `.tmp` staging file.
    LeaderEpochCheckpoint,
    /// The partition's `log-start-offset-checkpoint`, or its `.tmp` staging
    /// file.
    LogStartOffsetCheckpoint,
    /// A compaction `.swap` file, or a segment file the swap replaces.
    CompactionSwap,
    /// A segment file being renamed to its `.deleted` tombstone or unlinked
    /// by retention.
    SegmentDeletion,
    /// A file outside any partition log that another crate replaces through
    /// [`write_file_atomic`]. Only the real-file I/O ever writes one.
    External,
}

/// The operating-system I/O boundary every durable log write crosses.
///
/// The default methods perform real file I/O. Tests can override only the
/// operation they need to fail, while reads continue to use the segment's
/// shared `Arc<File>` directly.
pub trait LogIo: Debug + Send + Sync {
    /// Write bytes at the active `.log` file's current cursor.
    ///
    /// # Errors
    /// Returns the underlying write error.
    fn write(&self, file: &File, buf: &[u8]) -> io::Result<usize> { (&*file).write(buf) }

    /// Write byte slices at the active `.log` file's current cursor.
    ///
    /// # Errors
    /// Returns the underlying vectored-write error.
    fn write_vectored(&self, file: &File, bufs: &[IoSlice<'_>]) -> io::Result<usize> {
        (&*file).write_vectored(bufs)
    }

    /// Flush the active `.log` file's data to stable storage.
    ///
    /// # Errors
    /// Returns the underlying data-sync error.
    fn sync_data(&self, file: &File) -> io::Result<()> { file.sync_data() }

    /// Start the write-out of `len` bytes of `file`'s page-cache data from
    /// `offset`, without waiting for it and without a journal commit or a
    /// disk-cache flush. On Linux this is `sync_file_range(2)` with
    /// `SYNC_FILE_RANGE_WRITE`; elsewhere it is [`Self::sync_data`], which
    /// does more and blocks. `Log` paces the active segment's dirty data with
    /// it, so an `fdatasync` of the file, which stalls the appends that run
    /// while it is in progress, happens only at the roll and on request.
    ///
    /// # Errors
    /// Returns the underlying write-out error.
    #[cfg(not(target_os = "wasi"))]
    fn write_back(&self, file: &File, offset: u64, len: u64) -> io::Result<()> {
        write_back(file, offset, len)
    }

    /// Write bytes at `file`'s current cursor on behalf of `target`.
    ///
    /// Like [`std::io::Write::write`], this may write fewer bytes than asked
    /// for; [`write_all`] is the loop that finishes the buffer.
    ///
    /// # Errors
    /// Returns the underlying write error.
    fn write_at(&self, target: IoTarget, file: &File, buf: &[u8]) -> io::Result<usize> {
        let _ = target;
        (&*file).write(buf)
    }

    /// Flush `target`'s file data to stable storage.
    ///
    /// # Errors
    /// Returns the underlying data-sync error.
    fn sync_file(&self, target: IoTarget, file: &File) -> io::Result<()> {
        let _ = target;
        file.sync_data()
    }

    /// `fsync` a directory so the names created or renamed inside it are
    /// durable.
    ///
    /// # Errors
    /// Returns the underlying open or sync error.
    fn sync_dir(&self, dir: &Path) -> io::Result<()> {
        // Rust's standard directory-open path is supported on Unix, where
        // syncing the parent is what makes a rename or a fresh name durable.
        // Windows offers no equivalent through `std` (`File::open` on a
        // directory fails with `EACCES`), so the call is a no-op there and on
        // every other non-Unix target, `wasm32-wasip1` included. The file
        // contents are still synced before every rename.
        #[cfg(unix)]
        {
            File::open(dir)?.sync_all()
        }
        #[cfg(not(unix))]
        {
            let _ = dir;
            Ok(())
        }
    }

    /// Rename a file on behalf of `target`.
    ///
    /// # Errors
    /// Returns the underlying rename error.
    fn rename(&self, target: IoTarget, from: &Path, to: &Path) -> io::Result<()> {
        let _ = target;
        fs::rename(from, to)
    }

    /// Unlink a file on behalf of `target`.
    ///
    /// # Errors
    /// Returns the underlying unlink error.
    fn remove_file(&self, target: IoTarget, path: &Path) -> io::Result<()> {
        let _ = target;
        fs::remove_file(path)
    }

    /// Tell the kernel that a fetch reads `len` bytes of a `.log` file from
    /// `offset` next.
    ///
    /// This is the one read-side method, and it is a hint, not a read: it
    /// starts `posix_fadvise(POSIX_FADV_WILLNEED)` readahead of the range and
    /// returns without waiting for it. The real implementation issues the hint
    /// in 128 KiB calls, because Linux reads only the start of one large
    /// `WILLNEED` range. A failure changes nothing a reader can
    /// observe, so it is dropped. The targets without `posix_fadvise` do
    /// nothing.
    fn advise_will_need(&self, file: &File, offset: u64, len: u64) {
        advise_will_need(file, offset, len);
    }

    /// Drop the page cache's copy of the `len` bytes of a `.log` file from
    /// `offset`, once the kernel has written them out.
    ///
    /// This waits for the range's write-out with `sync_file_range`, then
    /// gives `posix_fadvise(POSIX_FADV_DONTNEED)`. The kernel frees only
    /// clean pages, so a drop of a range whose write-out had only been
    /// started, as the budget's are when the budget is smaller than the
    /// write-out in flight, would leave its pages cached and the ledger no
    /// longer counting them; the wait is what makes a drop a drop. On pages
    /// written out long ago, every drop of a sealed segment, it returns at
    /// once. No data is lost either way; a later read of the range comes
    /// from the disk. A failure changes nothing a reader can observe, so it
    /// is dropped. The targets without these calls do nothing, and
    /// `wasm32-wasip1`, which has no flush pool, has no caller. See
    /// `log::page_cache`.
    #[cfg(not(target_os = "wasi"))]
    fn drop_cached(&self, file: &File, offset: u64, len: u64) { drop_cached(file, offset, len); }

    /// Whether the page cache holds the byte at `offset` of `file`, asked
    /// without waiting for the disk. The fetch path asks it before it walks a
    /// run's batch headers one small read at a time: on bytes that are not in
    /// the page cache, each of those reads waits for the disk under the log
    /// lock, and one large read is faster. A host that cannot ask answers
    /// `None`, which keeps the walk and lets the segment stop probing.
    #[krabka_macros::sendfile_cfg]
    fn is_cached(&self, file: &File, offset: u64) -> Option<bool> { is_cached(file, offset) }

    /// Reserve disk blocks for the `len` bytes of a `.log` file from `offset`
    /// without changing its length, so the appends that later fill the range
    /// do not allocate.
    ///
    /// This is `fallocate(FALLOC_FL_KEEP_SIZE)`. A reservation changes no
    /// byte a reader can see, so a caller treats a failure as a segment that
    /// grows as it is written. The targets without it answer `Unsupported`.
    ///
    /// # Errors
    /// Returns the underlying `fallocate` error, or `Unsupported` off Linux.
    fn reserve(&self, file: &File, offset: u64, len: u64) -> io::Result<()> {
        reserve(file, offset, len)
    }

    /// Give back the blocks a [`LogIo::reserve`] left unused past a `.log`
    /// file's end, which is `len`.
    ///
    /// This truncates the file to the length it already has. ext4 and tmpfs
    /// free every block past the end on a truncate, even one that moves
    /// nothing, and that is the only way to reach blocks reserved past the
    /// end: ext4 ignores `FALLOC_FL_PUNCH_HOLE` at or past it. A filesystem
    /// that keeps them keeps them until retention deletes the segment.
    ///
    /// # Errors
    /// Returns the underlying truncate error.
    fn release(&self, file: &File, len: u64) -> io::Result<()> { file.set_len(len) }

    /// Open a second handle on a `.log` file that bypasses the page cache
    /// (`O_DIRECT`), with the alignment every write through it needs.
    ///
    /// The alignment comes from `statx(STATX_DIOALIGN)` (Linux 6.1) and from
    /// nowhere else: a guessed one that is wrong fails the append that first
    /// uses it, and a failed append takes the log directory offline. A
    /// kernel or filesystem that does not report one answers `Unsupported`,
    /// and the segment stays buffered. Both alignments are then raised to the
    /// filesystem block, up to 4 KiB, so that no write's offset, length or
    /// buffer falls inside a block the filesystem would have to lock, convert
    /// or zero apart.
    ///
    /// # Errors
    /// Returns `Unsupported` where `O_DIRECT` is unavailable or its alignment
    /// unknown, or the underlying `statx` or `open` error.
    fn open_direct(&self, path: &Path) -> io::Result<DirectFile> { open_direct(path) }

    /// Write `buf` at `offset` through a handle [`LogIo::open_direct`]
    /// opened. `buf`, `offset` and the length are aligned as that handle
    /// requires. Like [`std::io::Write::write`], this may write fewer bytes
    /// than asked for.
    ///
    /// # Errors
    /// Returns the underlying write error.
    fn write_direct(&self, file: &File, buf: &[u8], offset: u64) -> io::Result<usize> {
        write_direct(file, buf, offset)
    }

    /// Whether the direct writes of a segment under this I/O go to the
    /// shared `io_uring` submitter, which writes the file itself, rather than
    /// through [`LogIo::write_direct`] on the write threads.
    ///
    /// Only the real file I/O answers true. Every other implementation keeps
    /// the threads, so an override of [`LogIo::write_direct`] sees each write.
    fn submits_direct_writes(&self) -> bool { false }
}

/// A `.log` handle that bypasses the page cache, and what it requires of a
/// write: a buffer address aligned to `mem_align`, and an offset and a length
/// that are multiples of `block`.
#[derive(Debug)]
pub struct DirectFile {
    /// The `O_DIRECT` handle.
    pub file: File,
    /// Alignment of a write's buffer address, in bytes: at least what
    /// `O_DIRECT` requires, and every write keeps to it.
    pub mem_align: usize,
    /// Alignment of a write's offset and length, in bytes: at least what
    /// `O_DIRECT` requires, and every write keeps to it.
    pub block: usize,
}

/// The largest filesystem block [`write_alignment`] raises a direct write's
/// alignment to. It is the bound on the zeros past the last batch that
/// `docs/persisted_formats.md` gives for an active `preallocate` segment.
/// `stx_blksize` is the preferred I/O size, which is the block size on ext4
/// but can be an XFS stripe width or an NFS `wsize`: those are larger than
/// any block, and the cap keeps them from growing every write.
#[cfg(target_os = "linux")]
const MAX_FS_BLOCK: u32 = 4096;

/// The alignment a direct write keeps for one of the two `O_DIRECT` rules,
/// whose kernel minimum is `required`: the buffer address, or the file
/// offset and length. It is `required` raised to the filesystem block
/// `fs_block`, capped at [`MAX_FS_BLOCK`], when that is a power of two.
///
/// The kernel takes buffers and offsets aligned to the device's logical
/// block, often 512 bytes, but a filesystem allocates and converts whole
/// blocks of its own. ext4 sends a direct write down its unaligned path when
/// its offset, its length or its buffer address falls inside a 4 KiB block:
/// it takes the inode lock exclusively, waits for every other direct write
/// to the file, and zeroes the rest of a newly allocated partial block with
/// a write of its own. A segment's appends end anywhere, so with the kernel
/// minimum almost every write paid that, at about a fifth of the device's
/// rate. The larger of two powers of two is a multiple of both, so the
/// result still meets the kernel's rule.
#[cfg(target_os = "linux")]
fn write_alignment(required: u32, fs_block: u32) -> u32 {
    let fs_block = fs_block.min(MAX_FS_BLOCK);
    if fs_block.is_power_of_two() {
        required.max(fs_block)
    } else {
        required
    }
}

#[cfg(target_os = "linux")]
fn open_direct(path: &Path) -> io::Result<DirectFile> {
    use rustix::fs::{AtFlags, CWD, Mode, OFlags, StatxFlags};

    let stat = statx(CWD, path, AtFlags::empty(), StatxFlags::DIOALIGN)?;
    let reported = stat.stx_mask & StatxFlags::DIOALIGN.bits() != 0;
    // A zero offset alignment is the kernel saying this file cannot be
    // opened `O_DIRECT` at all.
    if !reported || stat.stx_dio_offset_align == 0 || stat.stx_dio_mem_align == 0 {
        return Err(ErrorKind::Unsupported.into());
    }
    let to_usize =
        |align: u32| usize::try_from(align).map_err(|_| io::Error::from(ErrorKind::Unsupported));
    let fd = open(
        path,
        OFlags::RDWR | OFlags::DIRECT | OFlags::CLOEXEC,
        Mode::empty(),
    )?;
    Ok(DirectFile {
        file: File::from(fd),
        mem_align: to_usize(write_alignment(stat.stx_dio_mem_align, stat.stx_blksize))?,
        block: to_usize(write_alignment(stat.stx_dio_offset_align, stat.stx_blksize))?,
    })
}

#[cfg(not(target_os = "linux"))]
fn open_direct(path: &Path) -> io::Result<DirectFile> {
    let _ = path;
    Err(ErrorKind::Unsupported.into())
}

#[cfg(unix)]
fn write_direct(file: &File, buf: &[u8], offset: u64) -> io::Result<usize> {
    FileExt::write_at(file, buf, offset)
}

/// No target without `O_DIRECT` ever opens a handle to write through.
#[cfg(not(unix))]
fn write_direct(file: &File, buf: &[u8], offset: u64) -> io::Result<usize> {
    let _ = (file, buf, offset);
    Err(ErrorKind::Unsupported.into())
}

#[cfg(target_os = "linux")]
fn reserve(file: &File, offset: u64, len: u64) -> io::Result<()> {
    fallocate(file, FallocateFlags::KEEP_SIZE, offset, len).map_err(Into::into)
}

#[cfg(not(target_os = "linux"))]
fn reserve(file: &File, offset: u64, len: u64) -> io::Result<()> {
    let _ = (file, offset, len);
    Err(ErrorKind::Unsupported.into())
}

/// The size of one `POSIX_FADV_WILLNEED` call.
///
/// Linux reads at most `max(max_sectors_kb, read_ahead_kb)` of one `WILLNEED`
/// call and drops the rest of the range. That is about 4 MiB on a local disk
/// and often less than 1 MiB on a cloud disk, so a hint over a whole window
/// reads only its start. 128 KiB is the default `read_ahead_kb`, so each call
/// of this size reads all of its range.
#[cfg(any(target_os = "linux", test))]
const WILL_NEED_CHUNK: u64 = 128 * 1024;

/// The `(offset, len)` ranges, each at most [`WILL_NEED_CHUNK`] long, that
/// cover the `len` bytes from `offset` in order. The range stops at
/// `u64::MAX` when `offset + len` is larger.
#[cfg(any(target_os = "linux", test))]
fn will_need_chunks(offset: u64, len: u64) -> impl Iterator<Item = (u64, u64)> {
    let end = offset.saturating_add(len);
    iter::successors(Some(offset), |start| {
        Some(start.saturating_add(WILL_NEED_CHUNK))
    })
    .take_while(move |start| *start < end)
    .map(move |start| (start, WILL_NEED_CHUNK.min(end - start)))
}

#[cfg(target_os = "linux")]
fn advise_will_need(file: &File, offset: u64, len: u64) {
    for (offset, len) in will_need_chunks(offset, len) {
        if let Some(len) = NonZeroU64::new(len) {
            // A refused hint leaves the read exactly as it was without one.
            let _ = fadvise(file, offset, Some(len), Advice::WillNeed);
        }
    }
}

#[cfg(not(target_os = "linux"))]
fn advise_will_need(file: &File, offset: u64, len: u64) { let _ = (file, offset, len); }

#[cfg(target_os = "linux")]
fn write_back(file: &File, offset: u64, len: u64) -> io::Result<()> {
    sync_file_range(file, offset, len, SyncFileRangeFlags::WRITE)?;
    Ok(())
}

#[cfg(not(any(target_os = "linux", target_os = "wasi")))]
fn write_back(file: &File, offset: u64, len: u64) -> io::Result<()> {
    let _ = (offset, len);
    file.sync_data()
}

#[cfg(target_os = "linux")]
fn drop_cached(file: &File, offset: u64, len: u64) {
    if let Some(len) = NonZeroU64::new(len) {
        // The pages must be clean to go: wait for a write-out still in
        // flight, and start one for a page that is still dirty. A failure
        // here leaves the hint to do what it can.
        let _ = sync_file_range(
            file,
            offset,
            len.get(),
            SyncFileRangeFlags::WAIT_BEFORE
                | SyncFileRangeFlags::WRITE
                | SyncFileRangeFlags::WAIT_AFTER,
        );
        // A refused hint leaves the pages as they were.
        let _ = fadvise(file, offset, Some(len), Advice::DontNeed);
    }
}

#[cfg(not(any(target_os = "linux", target_os = "wasi")))]
fn drop_cached(file: &File, offset: u64, len: u64) { let _ = (file, offset, len); }

/// `preadv2(RWF_NOWAIT)` of one byte: it fails with `EAGAIN` instead of
/// reading from the disk when the page cache does not hold the byte. Linux
/// 5.9 and 5.10 can return zero bytes instead (a bug `preadv2(2)` documents);
/// callers only ask about bytes below the end of the file, so zero bytes also
/// means not cached.
#[cfg(target_os = "linux")]
fn is_cached(file: &File, offset: u64) -> Option<bool> {
    let mut byte = [0u8; 1];
    match preadv2(
        file,
        &mut [std::io::IoSliceMut::new(&mut byte)],
        offset,
        ReadWriteFlags::NOWAIT,
    ) {
        Err(Errno::OPNOTSUPP | Errno::NOSYS) => None,
        Err(Errno::AGAIN) | Ok(0) => Some(false),
        _ => Some(true),
    }
}

#[cfg(not(target_os = "linux"))]
#[krabka_macros::sendfile_cfg]
fn is_cached(file: &File, offset: u64) -> Option<bool> {
    let _ = (file, offset);
    None
}

/// Write the whole of `buf` to `file` through `io`, looping over short writes.
///
/// A `LogIo` that writes part of the buffer and then fails leaves the file
/// torn at exactly that boundary, which is the shape a disk-full write has on
/// a real filesystem.
///
/// # Errors
/// Returns the first error the underlying writes produce, or `WriteZero` when
/// a write makes no progress.
pub fn write_all(io: &dyn LogIo, target: IoTarget, file: &File, buf: &[u8]) -> io::Result<()> {
    write_all_with(buf, |remaining| io.write_at(target, file, remaining))
}

/// Complete a scalar write, retrying interrupts and rejecting zero progress.
pub fn write_all_with(
    mut buf: &[u8],
    mut write: impl FnMut(&[u8]) -> io::Result<usize>,
) -> io::Result<()> {
    while !buf.is_empty() {
        let written = write_progress(|| write(buf))?;
        buf = buf.get(written..).ok_or_else(|| {
            io::Error::new(
                ErrorKind::InvalidData,
                "a write reported more bytes than it was given",
            )
        })?;
    }
    Ok(())
}

/// Retry an interrupted write and require it to make progress.
pub fn write_progress(mut write: impl FnMut() -> io::Result<usize>) -> io::Result<usize> {
    loop {
        match write() {
            Ok(0) => return Err(ErrorKind::WriteZero.into()),
            Err(error) if error.kind() == ErrorKind::Interrupted => {}
            result => return result,
        }
    }
}

/// Replace `path` with `bytes` atomically and durably through real file I/O:
/// write `tmp`, sync it, rename it over `path`, then sync the parent
/// directory.
///
/// This is the entry point for files outside a partition log, such
/// as the controller's `quorum-state` file and its metadata checkpoints, which
/// have no fault-injection seam of their own.
///
/// # Errors
/// Returns the first create, write, sync or rename error.
pub fn write_file_atomic(tmp: &Path, path: &Path, bytes: &[u8]) -> io::Result<()> {
    write_atomic(&FileIo, IoTarget::External, tmp, path, bytes, true)
}

/// Replace `path` with `bytes` atomically: write them to `tmp`, flush `tmp` to
/// stable storage, rename it over `path`, and, when `sync_dir` is set, `fsync`
/// `path`'s parent directory so the rename itself is durable.
///
/// Every step goes through `io` on behalf of `target`, so a fault injector can
/// fail the write, the flush, the rename or the directory sync. A failure
/// before the rename leaves `path` as it was; `tmp` may be left behind.
///
/// # Errors
/// Returns the first create, write, sync or rename error.
pub fn write_atomic(
    io: &dyn LogIo,
    target: IoTarget,
    tmp: &Path,
    path: &Path,
    bytes: &[u8],
    sync_dir: bool,
) -> io::Result<()> {
    {
        let file = File::create(tmp)?;
        write_all(io, target, &file, bytes)?;
        io.sync_file(target, &file)?;
    }
    io.rename(target, tmp, path)?;
    if sync_dir {
        let parent = path
            .parent()
            .filter(|parent| !parent.as_os_str().is_empty())
            .unwrap_or_else(|| Path::new("."));
        io.sync_dir(parent)?;
    }
    Ok(())
}

#[derive(Debug)]
pub struct FileIo;

impl LogIo for FileIo {
    fn submits_direct_writes(&self) -> bool { true }
}

/// The shared handle every structure defaults to before a test installs its
/// own.
pub fn file_io() -> Arc<dyn LogIo> { Arc::new(FileIo) }

#[cfg(test)]
mod tests {
    use {
        super::*,
        assert2::check,
        rustix::fs::fstatvfs,
        std::{
            fs,
            io::{self, ErrorKind},
            path::PathBuf,
            sync::Mutex,
            vec,
        },
    };

    /// A writer that replays a script of results, recording what each call was
    /// offered. A regular file cannot be persuaded to write short or to make
    /// no progress on demand, so the loop's two obligations -- resume where the
    /// last write stopped, and refuse to spin on a writer that writes nothing
    /// -- are only reachable through one of these.
    #[derive(Debug)]
    struct Scripted {
        script: Mutex<vec::IntoIter<io::Result<usize>>>,
        seen: Mutex<Vec<Vec<u8>>>,
    }

    impl Scripted {
        fn new(script: Vec<io::Result<usize>>) -> Self {
            Self {
                script: Mutex::new(script.into_iter()),
                seen: Mutex::new(Vec::new()),
            }
        }
    }

    impl LogIo for Scripted {
        fn write_at(&self, _target: IoTarget, _file: &File, buf: &[u8]) -> io::Result<usize> {
            self.seen.lock().unwrap().push(buf.to_vec());
            self.script
                .lock()
                .unwrap()
                .next()
                .unwrap_or(Ok(buf.len()))
                .map(|written| written.min(buf.len()))
        }
    }

    /// A file in a temporary directory.
    struct ScratchFile {
        dir: tempfile::TempDir,
        file: File,
    }

    fn scratch_file() -> ScratchFile {
        let dir = tempfile::tempdir().unwrap();
        let file = File::create(dir.path().join("scratch")).unwrap();
        ScratchFile { dir, file }
    }

    /// A state file with contents, and the staging path beside it, in a
    /// temporary directory.
    struct ExistingState {
        dir: tempfile::TempDir,
        path: PathBuf,
        tmp: PathBuf,
    }

    fn existing_state(contents: &[u8]) -> ExistingState {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state");
        let tmp = dir.path().join("state.tmp");
        fs::write(&path, contents).unwrap();
        ExistingState { dir, path, tmp }
    }

    #[test]
    fn will_need_chunks_cover_the_span_in_chunks() {
        const C: u64 = WILL_NEED_CHUNK;
        let cases = [
            ("empty", 4096, 0, vec![]),
            ("shorter than a chunk", 4096, 100, vec![(4096, 100)]),
            ("one chunk", 0, C, vec![(0, C)]),
            ("exact multiple", 10, 3 * C, vec![
                (10, C),
                (10 + C, C),
                (10 + 2 * C, C),
            ]),
            ("remainder", 0, 2 * C + 7, vec![(0, C), (C, C), (2 * C, 7)]),
            ("saturates at u64::MAX", u64::MAX - C - 5, 3 * C, vec![
                (u64::MAX - C - 5, C),
                (u64::MAX - 5, 5),
            ]),
            ("starts at u64::MAX", u64::MAX, C, vec![]),
        ];
        for (name, offset, len, expected) in cases {
            let chunks: Vec<_> = will_need_chunks(offset, len).collect();
            check!(chunks == expected, "{name}");
        }
    }

    #[test]
    fn write_atomic_replaces_the_target_and_consumes_the_staging_file() {
        let ExistingState {
            dir: _dir,
            path,
            tmp,
        } = existing_state(b"old contents");

        for contents in [b"new".as_slice(), b"newer"] {
            write_file_atomic(&tmp, &path, contents).unwrap();
            check!(fs::read(&path).unwrap() == contents.to_vec());
            check!(!tmp.exists());
        }
    }

    #[test]
    fn write_atomic_leaves_the_target_alone_when_the_staged_write_fails() {
        let ExistingState {
            dir: _dir,
            path,
            tmp,
        } = existing_state(b"old");

        let io = Scripted::new(vec![Err(io::Error::from(ErrorKind::StorageFull))]);
        let error = write_atomic(&io, IoTarget::External, &tmp, &path, b"new", true).unwrap_err();
        check!(error.kind() == ErrorKind::StorageFull);
        check!(fs::read(&path).unwrap() == b"old".to_vec());
    }

    #[test]
    fn write_all_resumes_short_writes_retries_interruptions_and_stops_on_no_progress() {
        let ScratchFile { dir: _dir, file } = scratch_file();

        for (script, expected) in [
            // Three short writes finish a six-byte buffer, each offered only what
            // the last one left.
            (vec![Ok(2), Ok(3), Ok(1)], vec![
                b"abcdef".to_vec(),
                b"cdef".to_vec(),
                b"f".to_vec(),
            ]),
            // An interrupted write is retried with the same bytes, not skipped past.
            (
                vec![Ok(2), Err(io::Error::from(ErrorKind::Interrupted)), Ok(4)],
                vec![b"abcdef".to_vec(), b"cdef".to_vec(), b"cdef".to_vec()],
            ),
        ] {
            let io = Scripted::new(script);
            write_all(&io, IoTarget::OffsetIndex, &file, b"abcdef").unwrap();
            check!(*io.seen.lock().unwrap() == expected);
        }
        for (script, expected) in [
            // A writer that makes no progress is a `WriteZero`, not a spin.
            (vec![Ok(0)], ErrorKind::WriteZero),
            // Any other error is returned as it stands.
            (
                vec![Ok(1), Err(io::Error::from(ErrorKind::StorageFull))],
                ErrorKind::StorageFull,
            ),
        ] {
            let io = Scripted::new(script);
            let error = write_all(&io, IoTarget::OffsetIndex, &file, b"abcdef").unwrap_err();
            check!(error.kind() == expected);
        }
    }

    /// A direct write keeps to the filesystem block as well as to what
    /// `O_DIRECT` requires, so neither its buffer nor its offset and length
    /// ever fall inside a block the filesystem allocates. A preferred I/O size
    /// above 4 KiB counts as 4 KiB, and one that is not a power of two below
    /// that, or none, leaves the `O_DIRECT` alignment as it is.
    #[cfg(target_os = "linux")]
    #[yare::parameterized(
        ext4_on_a_512_byte_device = { 512, 4096, 4096 },
        ext4_on_a_4k_device = { 4096, 4096, 4096 },
        small_filesystem_block = { 512, 1024, 1024 },
        device_coarser_than_the_filesystem = { 4096, 1024, 4096 },
        xfs_stripe_width = { 512, 65536, 4096 },
        nfs_wsize = { 512, 1_048_576, 4096 },
        stripe_width_not_a_power_of_two = { 512, 196_608, 4096 },
        small_block_not_a_power_of_two = { 512, 3072, 512 },
        no_block_reported = { 512, 0, 512 },
    )]
    fn a_direct_write_aligns_to_the_filesystem_block(required: u32, fs_block: u32, expected: u32) {
        check!(write_alignment(required, fs_block) == expected);
    }

    /// The real `O_DIRECT` handle, where the scratch filesystem offers one,
    /// asks for buffers and offsets aligned to the filesystem block, up to
    /// 4 KiB, whatever the device under it takes.
    #[cfg(target_os = "linux")]
    #[test]
    fn a_real_direct_handle_aligns_to_the_filesystem_block() {
        let ScratchFile { dir, file } = scratch_file();
        let Ok(direct) = open_direct(&dir.path().join("scratch")) else {
            return;
        };
        let fs_block = usize::try_from(fstatvfs(&file).unwrap().f_bsize.min(4096)).unwrap();
        for align in [direct.mem_align, direct.block] {
            check!(align.is_power_of_two());
            check!(align % fs_block == 0);
        }
    }
}