znippy-plugin-git 0.1.1

//! Impl 3 — the io_uring arm: **one submission, four ops, kernel-enforced
//! ordering**.
//!
//! [`SafeWriter`](crate::archive_write::SafeWriter) buys its durability with
//! four blocking syscalls and four round-trips through the scheduler:
//!
//! ```text
//!   pwrite(blob)  fsync(blob)  write(journal)  fsync(journal)
//!      ↑ ring 0      ↑ ring 0       ↑ ring 0        ↑ ring 0     4 syscalls
//! ```
//!
//! Here the same four operations are **one** `io_uring_enter(2)`. Each of the
//! first three SQEs carries `IOSQE_IO_LINK`, so the kernel will not start an op
//! until its predecessor has completed — the ordering contract is enforced by
//! the kernel rather than by the caller blocking between calls:
//!
//! ```text
//!   [ Write(blob) ]→[ Fsync(blob) ]→[ WriteFixed(journal) ]→[ Fsync(journal) ]
//!     IO_LINK         IO_LINK          IO_LINK                (chain end)
//!   ────────────────────── one io_uring_enter ──────────────────────
//! ```
//!
//! That is **the same ordering as `SafeWriter`**, not a second one: blob bytes
//! durable before the row that references them. A crash anywhere in the chain
//! leaves orphan payload nobody points at. If a link fails the kernel cancels
//! the rest of the chain with `ECANCELED`, which is exactly the semantics
//! wanted: no journal row is ever written after a failed blob fsync.
//!
//! # Registered buffers, and the copy that is *not* being avoided
//!
//! `register_buffers` pins a set of stable buffers once, so per-op the kernel
//! skips `get_user_pages`/`put_page` on them. That only works for buffers whose
//! address the ring can be told about **up front**, which the caller's transient
//! `&[u8]` is not. So this writer splits the two:
//!
//! * **journal** → a registered one-page staging buffer and `WriteFixed`. This is
//!   the buffer that is stable across appends, and it is where the win is real.
//!   Crucially it means there is **no arrow `BufWriter`** here at all: the
//!   serialized IPC message is written into the registered buffer and the kernel
//!   is handed that. You cannot have both arrow's buffered `StreamWriter` and a
//!   registered buffer; the brief says take the io_uring side, and this does.
//! * **blob** → plain `Write` against the caller's own pointer. Not registered
//!   (it cannot be), but also **not copied** in userspace: the pushed pack bytes
//!   go from the caller's slice straight into the ring.
//!
//! Saying this plainly matters: the registered buffer removes per-op page
//! pinning on the journal, not a `memcpy` on the pack.
//!
//! # What a registration COSTS, and why the staging buffer is one page
//!
//! `IORING_REGISTER_BUFFERS` pins its pages against **`RLIMIT_MEMLOCK`**, and
//! the kernel charges them to `user->locked_vm` — a counter kept on the
//! `user_struct`, so it is **per-UID and shared by every process that user is
//! running**, not per-process and not per-ring
//! (`io_uring/rsrc.c: io_account_mem -> __io_account_mem`).
//!
//! Every store gets its own writer, therefore its own ring, therefore its own
//! registration. So the pinned pages are `stores × ceil(JOURNAL_STAGING /
//! PAGE_SIZE)` and the ceiling is a **hard, shared, uid-wide** one. The default
//! on oden (and on stock Debian/Ubuntu) is 8 MiB.
//!
//! MEASURED on oden 2026-08-14, `RLIMIT_MEMLOCK` 8 MiB, one fresh process per
//! reading, three readings each, identical code with only the buffer size
//! varied:
//!
//! ```text
//!   registered per ring   rings that fit        what refused the next one
//!   ───────────────────   ──────────────        ─────────────────────────
//!   64 KiB (16 pages)     87, 88, 87            IORING_REGISTER_BUFFERS
//!    4 KiB  (1 page)      397, 422, 419         io_uring_setup
//! ```
//!
//! with
//!
//! ```text
//!   uring: register_buffers: Cannot allocate memory (os error 12)
//! ```
//!
//! and every store opened after that refused. **~87 io_uring stores per
//! process**, and gunnar's `multiuser_scaling` alone provisions 102 users with a
//! repository each; a `gunnar serve` holds one store — therefore one ring,
//! therefore one registration — per repository it has open. `FastWriter` and
//! `SafeWriter` register nothing and have no such ceiling, which is exactly the
//! shape the sweep showed: four workloads and eight forge arms red on **both**
//! io_uring columns and green on `fast` and `safe`, with the index arm varied
//! underneath and making no difference.
//!
//! Cutting the registration to one page moves that to **~400 rings, a 4.6×
//! ceiling** — and past it the thing that refuses is no longer the registration
//! at all but `io_uring_setup`. A bare ring of [`RING_ENTRIES`] costs about 21 KiB
//! of the same budget (8 MiB / ~400), so the old writer spent ~21 pages per store
//! and the new one spends ~6. It is not the 16× the buffer sizes suggest, because
//! the ring was always paying five pages of it.
//!
//! End to end rather than at the writer: **150 whole `GitStore`s on this arm,
//! open at once in one process**, each with a real 5 653 302-byte / 2 687-object
//! pack pushed, indexed and read back — 3.4–4.0 s per store, oden 2026-08-14,
//! load average 15–30. Every one of them past the old ceiling.
//!
//! # …and every number in the two paragraphs above was measured in the WRONG
//! # PROCESS: the page you register is charged as the huge page it sits in
//!
//! *Found 2026-08-14, after the ceiling above had already been "fixed" once.*
//!
//! **`io_buffer_account_pin` does not charge the pages you named. It charges the
//! `compound_head` of each of them.** If the 4 KiB you register happens to live
//! inside a transparent huge page, the kernel charges **the whole 2 MiB — 512
//! pages, not one** (`io_uring/rsrc.c: io_buffer_account_pin`, the
//! `PageCompound` branch: `imu->acct_pages += page_size(hpage) >> PAGE_SHIFT`).
//!
//! MEASURED on oden 2026-08-14, `RLIMIT_MEMLOCK` 8 MiB, `transparent_hugepage
//! = [madvise]`, one registration of exactly one page, charge read back off the
//! kernel by binary-searching what could still be registered afterwards:
//!
//! ```text
//!   where the one page came from                       pages charged
//!   ────────────────────────────────────────────────   ─────────────
//!   its own anonymous mmap, MADV_NOHUGEPAGE                        1
//!   glibc malloc(4096) in a small C program                        2
//!   inside a MADV_HUGEPAGE arena                                 512
//! ```
//!
//! `Box<[u8]>` — what this file registered until now — is *whatever the process
//! allocator gives you*, and **`gunnar serve` runs on mimalloc**, which
//! `madvise(MADV_HUGEPAGE)`s its arenas. Measured on the running server:
//! `AnonHugePages: 16384 kB` in one VMA, and every staging buffer allocated out
//! of it. So the real ceiling in the process that matters was not ~400 stores.
//! It was **three**:
//!
//! ```text
//!   RLIMIT_MEMLOCK    io_uring stores one `gunnar serve` could open
//!   ──────────────    ─────────────────────────────────────────────
//!   8 MiB                             3   (+ the control store = 4 × 2 MiB)
//!   4 MiB                             1
//! ```
//!
//! — measured by pushing to distinct repositories one at a time until the server
//! refused, oden 2026-08-14. Four huge pages fit in 8 MiB and that is the whole
//! arithmetic. The fourth push onwards died with
//! `register_buffers (4096 B, 1 page(s)): Cannot allocate memory`, which reads
//! like the ceiling this file already documents and is a different one: it is not
//! how MANY pages are registered, it is WHOSE page each one is.
//!
//! That is why [`Staging`] does not ask the allocator for the buffer. It takes
//! **its own one-page anonymous mapping and `madvise(MADV_NOHUGEPAGE)`s it**, so
//! the registration is charged one page whatever the executable's allocator does
//! — and the "~400 rings" arithmetic above becomes true instead of merely
//! plausible. `enough_uring_writers_for_a_population_coexist_in_one_process`
//! could never have caught this: a `cargo test` binary is on the system
//! allocator, whose 4 KiB allocations are not huge-page backed, so the guard
//! measured a process that did not have the bug. The guard that does catch it is
//! [`tests::a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in`],
//! which measures the CHARGE rather than the count.
//!
//! # It is tighter than "N stores at once", because the kernel reclaims lazily
//!
//! `io_uring` teardown runs off a workqueue after the ring's last descriptor
//! closes, so the pages of a **dropped** writer stay charged for a while. Opening
//! and dropping one at a time and holding *nothing*, the 64 KiB writer was
//! refused at the **63rd**. A server that opens a store per repository and lets
//! it go — gunnar's `store_cache` map holds `Weak`s, so that is its shape while
//! seeding — therefore hits this after about sixty repositories, which is where
//! the sweep's `vs_forge_*` arms died seeding.
//!
//! The accounting is on the `user_struct`, so it also crosses process
//! boundaries: a fresh process was refused its **second** ring while a previous
//! test process's rings were still being torn down. That is also why there is no
//! guard on the serial shape — one written against it failed at 54 of 160 purely
//! on the residue of the guard that ran before it, and a test whose verdict
//! depends on what else the box did in the last few seconds is not a guard. The
//! serial ceiling is bounded by the same per-store cost the guards below do
//! measure, and it clears with pacing: 600 rings created and dropped 5 ms apart
//! did not fail once.
//!
//! # If the kernel cannot do it
//!
//! [`UringWriter::create`] probes for `IORING_OP_WRITE`, `IORING_OP_WRITE_FIXED`
//! and `IORING_OP_FSYNC` and fails with a named error if any is missing, rather
//! than silently degrading to `pwrite` and reporting an io_uring number that is
//! not one. Same for `io_uring_setup` being blocked outright
//! (`kernel.io_uring_disabled=2`, seccomp, a container without the syscall).

#![cfg(target_os = "linux")]

use std::fs::File;
use std::os::unix::fs::FileExt;
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};

use anyhow::{Result, anyhow, bail};
use io_uring::{IoUring, opcode, squeue, types};

use crate::archive_write::{
    ArchiveWrite, Extent, encode_journal_row, encode_journal_schema, journal_path, open_blobs,
};

/// Size of the registered staging buffer the journal message is built in.
///
/// **One page, because a registration is pinned memory charged uid-wide against
/// `RLIMIT_MEMLOCK`** — see the module docs for the arithmetic and the measured
/// failure. A page is the smallest thing the kernel can pin, so this is the
/// floor, and it is not tight: an encoded journal row is two `u64`s plus Arrow
/// IPC framing — **224 bytes**, measured by `encode_journal_row` on oden
/// 2026-08-14 — so one page is still 18× headroom. A row that ever outgrew it
/// would be refused by name in [`UringWriter::append`] rather than silently
/// truncated.
///
/// It was 64 KiB until 2026-08-14. That is 16 pages per store on top of the
/// ring's own ~5, and it is what capped a process at ~87 io_uring stores.
///
/// **The size is only half of it.** What the kernel charges depends on where the
/// page came from, not only on how many there are — see [`Staging`] and the
/// module docs' second cost section. One page out of mimalloc's huge-page arena
/// is charged 512.
const JOURNAL_STAGING: usize = 4096;

/// Largest single `Write` SQE. A blob bigger than this is split across several
/// linked `Write`s in the *same* chain, so the ordering guarantee is unchanged.
const MAX_WRITE_CHUNK: usize = 1 << 30; // 1 GiB

/// Ring depth. Enough for 61 blob chunks (61 GiB) plus fsync/journal/fsync.
const RING_ENTRIES: u32 = 64;

/// This kernel's page size — the granularity a registration is pinned at, so
/// the unit [`JOURNAL_STAGING`] is really measured in.
pub(crate) fn page_size() -> usize {
    // SAFETY: `sysconf` takes no pointers and cannot fail destructively; a
    // negative return (it has none for `_SC_PAGESIZE` on Linux) falls back to
    // the architectural 4 KiB.
    let n = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
    if n > 0 { n as usize } else { 4096 }
}

/// The soft `RLIMIT_MEMLOCK` in bytes, or `None` if it is unlimited or
/// unreadable.
///
/// This is the ceiling every `IORING_REGISTER_BUFFERS` in the process is
/// charged against, and the kernel keeps the running total on the
/// **`user_struct`** — so it is shared with every other process this uid is
/// running. That is why an io_uring store can be refused by a ring some
/// unrelated process of the same user opened.
pub(crate) fn memlock_limit_bytes() -> Option<u64> {
    let mut lim = libc::rlimit {
        rlim_cur: 0,
        rlim_max: 0,
    };
    // SAFETY: `lim` is a live, correctly-typed `rlimit` this call only writes.
    if unsafe { libc::getrlimit(libc::RLIMIT_MEMLOCK, &mut lim) } != 0 {
        return None;
    }
    if lim.rlim_cur == libc::RLIM_INFINITY {
        None
    } else {
        Some(lim.rlim_cur as u64)
    }
}

/// The registered staging buffer: **its own anonymous mapping, one page,
/// `MADV_NOHUGEPAGE`** — never the process allocator's memory.
///
/// It is a mapping rather than a `Box<[u8]>` for one reason, and it is the
/// module docs' second "what a registration costs" section: the kernel charges
/// a registration by `compound_head`, so a page handed out by an allocator that
/// `madvise(MADV_HUGEPAGE)`s its arenas — mimalloc, which `gunnar serve` runs on
/// — is charged as **512 pages**. Its own mapping cannot be part of anybody's
/// huge page, and `MADV_NOHUGEPAGE` says so to `khugepaged` as well rather than
/// relying on a one-page VMA being too small to collapse.
///
/// The address is stable for the mapping's whole life, which is what the
/// registration requires, and it is unmapped in [`Drop`] **after** the ring that
/// registered it has been dropped (field order in [`Ring`]).
struct Staging {
    ptr: *mut u8,
    len: usize,
}

impl Staging {
    /// Map `len` bytes, rounded up to whole pages, refusing huge pages.
    fn map(len: usize) -> Result<Self> {
        let page = page_size();
        let len = len.next_multiple_of(page).max(page);
        // SAFETY: a fresh anonymous mapping; no pointer of ours is passed in.
        let ptr = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                len,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
                -1,
                0,
            )
        };
        if ptr == libc::MAP_FAILED {
            return Err(anyhow!(
                "uring: mmap {len} B for the registered staging buffer: {}",
                std::io::Error::last_os_error()
            ));
        }
        // SAFETY: `ptr`/`len` are the mapping just returned.
        let advised = unsafe { libc::madvise(ptr, len, libc::MADV_NOHUGEPAGE) };
        if advised != 0 {
            // ENOSYS/EINVAL means this kernel has no transparent huge pages at
            // all, which is the state the advice was asking for. Anything else
            // is a refusal to give it, and the registration would then be
            // charged 512 pages instead of one — a silent 512× ceiling is
            // exactly what this file was fixed for, so it is named, not
            // swallowed.
            let e = std::io::Error::last_os_error();
            let benign = matches!(
                e.raw_os_error(),
                Some(libc::EINVAL) | Some(libc::ENOSYS)
            );
            if !benign {
                // SAFETY: unmapping the mapping made three statements ago.
                unsafe { libc::munmap(ptr, len) };
                return Err(anyhow!(
                    "uring: madvise(MADV_NOHUGEPAGE) on the staging buffer: {e}. Without it a \
                     one-page registration can be charged as the whole 2 MiB huge page it sits \
                     in, which caps a process at four io_uring stores on a stock 8 MiB \
                     RLIMIT_MEMLOCK."
                ));
            }
        }
        Ok(Self {
            ptr: ptr as *mut u8,
            len,
        })
    }

    fn len(&self) -> usize {
        self.len
    }
    fn as_ptr(&self) -> *const u8 {
        self.ptr
    }
    fn as_mut_slice(&mut self) -> &mut [u8] {
        // SAFETY: `ptr..ptr+len` is our own live, readable, writable mapping,
        // and `&mut self` is the only handle to it.
        unsafe { std::slice::from_raw_parts_mut(self.ptr, self.len) }
    }
}

impl Drop for Staging {
    fn drop(&mut self) {
        // SAFETY: the mapping this struct owns, unmapped exactly once. The ring
        // that registered it is dropped first (field order in `Ring`), so no
        // registration references these pages any more.
        unsafe { libc::munmap(self.ptr as *mut libc::c_void, self.len) };
    }
}

struct Ring {
    ring: IoUring,
    /// Registered index 0. Its own mapping, so its address is stable for the
    /// lifetime of the registration even though `UringWriter` may move — and so
    /// the kernel charges it one page. See [`Staging`].
    staging: Staging,
    journal_cursor: u64,
}

// SAFETY: every access to `Ring` goes through `UringWriter`'s `Mutex`, so the
// submission and completion queues are never touched from two threads at once.
// The registered buffer is owned here and outlives the registration (the ring is
// dropped first, in field order).
unsafe impl Send for Ring {}

/// io_uring writer: write → fsync → journal → fsync as one linked submission.
///
/// # Durability on return
///
/// Identical to [`SafeWriter`](crate::archive_write::SafeWriter): the pack bytes
/// are on the device and a journal row on the device references them. What
/// differs is the cost of getting there — one `io_uring_enter` instead of four
/// blocking syscalls — not what is promised.
pub struct UringWriter {
    blobs: File,
    journal: File,
    journal_file_path: PathBuf,
    cursor: AtomicU64,
    ring: Mutex<Ring>,
}

impl UringWriter {
    /// Open `archive`, set up the ring, register the journal staging buffer, and
    /// open the journal beside it — **appending** to the rows already there, and
    /// emitting the Arrow IPC schema header only if the file is new.
    ///
    /// Returns a named error — never a silent fallback — if this kernel cannot
    /// provide what the chain needs. Nothing on disk is touched before that
    /// point is passed.
    pub fn create(archive: &Path) -> Result<Self> {
        let (blobs, end) = open_blobs(archive)?;
        let jpath = journal_path(archive);

        // **The ring first, the journal second.** Nothing here may touch the
        // journal until this writer is certain it can be built: `create` can
        // still fail on `register_buffers` (the uid-wide `RLIMIT_MEMLOCK`
        // ceiling in the module docs), and a failed open that had already
        // opened the journal would leave a store's durable ack log standing
        // behind a writer that does not exist.
        let ring = IoUring::new(RING_ENTRIES).map_err(|e| {
            anyhow!(
                "uring: io_uring_setup failed ({e}). This kernel cannot run the io_uring arm \
                 (check /proc/sys/kernel/io_uring_disabled, seccomp, container policy). \
                 No fallback is substituted — a pwrite number reported as an io_uring number \
                 would be a lie."
            )
        })?;

        let mut probe = io_uring::register::Probe::new();
        ring.submitter()
            .register_probe(&mut probe)
            .map_err(|e| anyhow!("uring: register_probe: {e}"))?;
        for (code, what) in [
            (opcode::Write::CODE, "IORING_OP_WRITE"),
            (opcode::WriteFixed::CODE, "IORING_OP_WRITE_FIXED"),
            (opcode::Fsync::CODE, "IORING_OP_FSYNC"),
        ] {
            if !probe.is_supported(code) {
                bail!(
                    "uring: this kernel does not support {what}; the write→fsync→journal→fsync \
                     chain cannot be built. Not degrading to pwrite."
                );
            }
        }

        let staging = Staging::map(JOURNAL_STAGING)?;
        // SAFETY: `staging` owns its own mapping and is owned by the `Ring`
        // below; it is neither moved nor unmapped until the ring is dropped,
        // which happens before it (struct field order in `Ring`).
        unsafe {
            let iov = libc::iovec {
                iov_base: staging.as_ptr() as *mut libc::c_void,
                iov_len: staging.len(),
            };
            ring.submitter()
                .register_buffers(std::slice::from_ref(&iov))
                .map_err(|e| {
                    anyhow!(
                        "uring: register_buffers ({} B, {} page(s)): {e}. A registration is \
                         pinned memory charged against RLIMIT_MEMLOCK, and the kernel keeps that \
                         count on the user_struct — it is per-UID and shared with every other \
                         process this user is running, not per-process. This one is currently \
                         {}. Every io_uring store holds one registration for as long as it is \
                         open, so N stores pin N pages; `fast` and `safe` register nothing and \
                         have no such ceiling. The page(s) named here are what was ASKED for; \
                         the kernel charges by compound_head, so a staging buffer that ended up \
                         inside a transparent huge page would be charged the whole 2 MiB — see \
                         `Staging`, which takes its own MADV_NOHUGEPAGE mapping so that cannot \
                         happen.",
                        JOURNAL_STAGING,
                        JOURNAL_STAGING.div_ceil(page_size()),
                        memlock_limit_bytes()
                            .map(|b| format!("{b} B"))
                            .unwrap_or_else(|| "unreadable".into()),
                    )
                })?;
        }

        // **The journal is a LOG and a reopen APPENDS to it** — the contract
        // `archive_write`'s module docs state for every durable arm, and until
        // 2026-08-14 this writer was the one that broke it: it opened with
        // `File::create`, which truncates, so a store reopened on this arm
        // erased the durable record that its packs had ever been acked, derived
        // an empty crash-recovery diff, and restarted pack ordinals at 0 — see
        // `indexer::packs_already_acked` for what a restarted ordinal does to a
        // pack's objects. `truncate(false)` plus a cursor taken from the file's
        // own length is what makes the two durable arms actually
        // indistinguishable on disk, which is what LAW 5 already claimed of
        // them.
        let journal = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&jpath)
            .map_err(|e| anyhow!("uring: open journal {}: {e}", jpath.display()))?;
        let already = journal
            .metadata()
            .map_err(|e| anyhow!("uring: stat journal {}: {e}", jpath.display()))?
            .len();
        // The schema message opens the stream and is written **once per file**,
        // not once per writer: a second one mid-file makes arrow's
        // `StreamReader` — and therefore `read_journal` — stop at it.
        let journal_cursor = if already == 0 {
            let schema = encode_journal_schema()?;
            journal.write_all_at(&schema, 0)?;
            journal.sync_all()?;
            schema.len() as u64
        } else {
            already
        };

        Ok(Self {
            blobs,
            journal,
            journal_file_path: jpath,
            cursor: AtomicU64::new(end),
            ring: Mutex::new(Ring {
                ring,
                staging,
                journal_cursor,
            }),
        })
    }

    /// Path of the journal segment beside `archive`.
    pub fn journal_path(archive: &Path) -> PathBuf {
        journal_path(archive)
    }

    /// The blob file, for a reader that wants to `pread` an extent back.
    pub fn blobs(&self) -> &File {
        &self.blobs
    }

    /// The journal file path this writer is appending rows to.
    pub fn journal_file(&self) -> &Path {
        &self.journal_file_path
    }
}

impl ArchiveWrite for UringWriter {
    fn append(&self, bytes: &[u8]) -> Result<Extent> {
        let len = bytes.len() as u64;
        let offset = self.cursor.fetch_add(len, Ordering::SeqCst);

        let chunks = bytes.len().div_ceil(MAX_WRITE_CHUNK).max(1);
        if chunks + 3 > RING_ENTRIES as usize {
            bail!(
                "uring: {} B needs {chunks} linked writes, more than the ring's {RING_ENTRIES} \
                 entries",
                bytes.len()
            );
        }

        let row = encode_journal_row(offset, len)?;
        let mut g = self
            .ring
            .lock()
            .map_err(|_| anyhow!("uring mutex poisoned"))?;
        if row.len() > g.staging.len() {
            bail!(
                "uring: journal row is {} B, staging buffer is {} B",
                row.len(),
                g.staging.len()
            );
        }
        g.staging.as_mut_slice()[..row.len()].copy_from_slice(&row);
        let journal_at = g.journal_cursor;

        let blob_fd = types::Fd(self.blobs.as_raw_fd());
        let journal_fd = types::Fd(self.journal.as_raw_fd());
        let staging_ptr = g.staging.as_ptr();

        let mut sqes: Vec<squeue::Entry> = Vec::with_capacity(chunks + 3);
        // 1..=chunks — the pack bytes, verbatim, straight from the caller's slice.
        for c in 0..chunks {
            let start = c * MAX_WRITE_CHUNK;
            let n = (bytes.len() - start).min(MAX_WRITE_CHUNK);
            sqes.push(
                opcode::Write::new(blob_fd, unsafe { bytes.as_ptr().add(start) }, n as u32)
                    .offset(offset + start as u64)
                    .build()
                    .flags(squeue::Flags::IO_LINK)
                    .user_data(c as u64),
            );
        }
        // chunks+1 — blob bytes durable BEFORE the row that references them.
        sqes.push(
            opcode::Fsync::new(blob_fd)
                .build()
                .flags(squeue::Flags::IO_LINK)
                .user_data(0xF5_00),
        );
        // chunks+2 — the journal row, out of the registered buffer, no BufWriter.
        sqes.push(
            opcode::WriteFixed::new(journal_fd, staging_ptr, row.len() as u32, 0)
                .offset(journal_at)
                .build()
                .flags(squeue::Flags::IO_LINK)
                .user_data(0x30_01),
        );
        // chunks+3 — the row is durable. Chain end: no IO_LINK.
        sqes.push(opcode::Fsync::new(journal_fd).build().user_data(0xF5_01));

        let want = sqes.len();
        // SAFETY: every buffer referenced by an SQE (`bytes`, `g.staging`)
        // outlives the `submit_and_wait` below, which does not return until all
        // `want` operations have completed. The fds outlive `self`.
        unsafe {
            g.ring
                .submission()
                .push_multiple(&sqes)
                .map_err(|e| anyhow!("uring: submission queue full: {e}"))?;
        }
        g.ring
            .submit_and_wait(want)
            .map_err(|e| anyhow!("uring: io_uring_enter: {e}"))?;

        let mut written = 0i64;
        let mut seen = 0usize;
        let mut journal_written = 0i64;
        for cqe in g.ring.completion() {
            seen += 1;
            let res = cqe.result();
            if res < 0 {
                let e = std::io::Error::from_raw_os_error(-res);
                bail!(
                    "uring: op {:#x} failed: {e} (a linked chain cancels its tail with \
                     ECANCELED, so no journal row was written after a failed blob fsync)",
                    cqe.user_data()
                );
            }
            match cqe.user_data() {
                0xF5_00 | 0xF5_01 => {}
                0x30_01 => journal_written = res as i64,
                _ => written += res as i64,
            }
        }
        if seen != want {
            bail!("uring: expected {want} completions, saw {seen}");
        }
        if written as u64 != len {
            bail!(
                "uring: short write — asked for {len} B, kernel wrote {written} B (io_uring \
                 Write is not write_all)"
            );
        }
        if journal_written as usize != row.len() {
            bail!(
                "uring: short journal write — {} B of {} B",
                journal_written,
                row.len()
            );
        }
        g.journal_cursor = journal_at + row.len() as u64;

        Ok((offset, len))
    }

    fn name(&self) -> &'static str {
        "UringWriter"
    }

    fn durability(&self) -> &'static str {
        "full — kernel-ordered blob fsync then journal fsync, one io_uring_enter"
    }
}


#[cfg(test)]
mod tests {
    use super::*;
    use crate::archive_write::{ArchiveWrite, read_journal};

    fn loadavg() -> String {
        std::fs::read_to_string("/proc/loadavg")
            .unwrap_or_default()
            .split_whitespace()
            .take(3)
            .collect::<Vec<_>>()
            .join(" ")
    }

    // ── the instrument the charge guard is built on ─────────────────────────
    //
    // `user->locked_vm` is not readable from userspace, so the only honest way
    // to ask what a registration COST is to ask the kernel what can still be
    // registered afterwards. `probe` is one long-lived ring; `scratch` is one
    // big NOHUGEPAGE mapping. Registering and *explicitly* unregistering
    // releases the charge synchronously — unlike dropping a ring, whose
    // un-accounting runs off a workqueue — so the measurement leaves no residue
    // of its own.

    /// **The two guards that measure `RLIMIT_MEMLOCK` may not run at once.**
    ///
    /// `cargo test` runs them on different threads of ONE process and the budget
    /// is a single uid-wide counter, so they are not independent measurements of
    /// anything: [`Headroom::pages`] transiently registers the whole limit to
    /// find out what is left, which is indistinguishable — to
    /// [`enough_uring_writers_for_a_population_coexist_in_one_process`] — from
    /// the ceiling it exists to detect. Seen: that guard failing at writer 84 of
    /// 160 purely because the charge guard was mid-binary-search beside it.
    ///
    /// This does not pretend to serialise other PROCESSES; the module docs
    /// already say the budget is shared uid-wide and that no guard here can own
    /// it. It removes the one source of interference that is ours.
    static MEMLOCK_MEASUREMENT: Mutex<()> = Mutex::new(());

    fn measuring_memlock() -> std::sync::MutexGuard<'static, ()> {
        MEMLOCK_MEASUREMENT
            .lock()
            .unwrap_or_else(|p| p.into_inner())
    }

    /// Big enough to ask for the whole stock 8 MiB limit in one registration.
    const HEADROOM_CAP_PAGES: usize = 2048;

    struct Headroom {
        probe: IoUring,
        scratch: Staging,
    }

    impl Headroom {
        fn new() -> Self {
            Self {
                probe: IoUring::new(8).expect("a probe ring"),
                scratch: Staging::map(HEADROOM_CAP_PAGES * page_size())
                    .expect("a scratch mapping"),
            }
        }
        fn fits(&self, pages: usize) -> bool {
            let iov = libc::iovec {
                iov_base: self.scratch.as_ptr() as *mut libc::c_void,
                iov_len: pages * page_size(),
            };
            // SAFETY: `iov` names our own live scratch mapping, and the
            // registration is dropped again before this function returns.
            let ok = unsafe {
                self.probe
                    .submitter()
                    .register_buffers(std::slice::from_ref(&iov))
            }
            .is_ok();
            if ok {
                self.probe
                    .submitter()
                    .unregister_buffers()
                    .expect("unregister the probe buffer");
            }
            ok
        }
        /// Largest registration, in pages, this uid could make right now.
        fn pages(&self) -> usize {
            if self.fits(HEADROOM_CAP_PAGES) {
                return HEADROOM_CAP_PAGES;
            }
            let (mut lo, mut hi) = (0usize, HEADROOM_CAP_PAGES);
            while lo + 1 < hi {
                let mid = (lo + hi) / 2;
                if self.fits(mid) { lo = mid } else { hi = mid }
            }
            lo
        }
    }

    /// A 2 MiB-aligned arena the kernel really has backed with a transparent
    /// huge page, or `None` if this box will not give one.
    struct HugeArena {
        raw: *mut libc::c_void,
        raw_len: usize,
        arena: *mut u8,
    }

    impl HugeArena {
        const HUGE: usize = 2 << 20;
        fn new() -> Option<Self> {
            let raw_len = 2 * Self::HUGE;
            // SAFETY: a fresh anonymous mapping.
            let raw = unsafe {
                libc::mmap(
                    std::ptr::null_mut(),
                    raw_len,
                    libc::PROT_READ | libc::PROT_WRITE,
                    libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
                    -1,
                    0,
                )
            };
            if raw == libc::MAP_FAILED {
                return None;
            }
            let arena = ((raw as usize + Self::HUGE - 1) & !(Self::HUGE - 1)) as *mut u8;
            // SAFETY: `arena..arena+HUGE` is inside the mapping above.
            unsafe {
                if libc::madvise(arena as *mut libc::c_void, Self::HUGE, libc::MADV_HUGEPAGE) != 0 {
                    libc::munmap(raw, raw_len);
                    return None;
                }
                // Fault it in, which is when the huge page is actually formed.
                std::ptr::write_bytes(arena, 1u8, Self::HUGE);
            }
            Some(Self {
                raw,
                raw_len,
                arena,
            })
        }
        /// A single page well inside the arena.
        fn a_page(&self) -> *const u8 {
            // SAFETY: four pages in, still inside the 2 MiB arena.
            unsafe { self.arena.add(4 * page_size()) }
        }
    }

    impl Drop for HugeArena {
        fn drop(&mut self) {
            // SAFETY: the mapping this struct owns.
            unsafe { libc::munmap(self.raw, self.raw_len) };
        }
    }

    /// **A registration is charged one page — not the 2 MiB huge page the page
    /// happens to live in.**
    ///
    /// This is the guard for the defect that survived the first fix of this
    /// file's ceiling and was found on 2026-08-14: `io_buffer_account_pin`
    /// charges by `compound_head`, so one 4 KiB registration taken out of an
    /// allocator arena that has been `madvise(MADV_HUGEPAGE)`d costs **512
    /// pages**. `gunnar serve` runs on mimalloc, which does exactly that, and
    /// the result was a server that could open **three** io_uring stores on a
    /// stock 8 MiB `RLIMIT_MEMLOCK` — four huge pages, one of them the control
    /// store — while [`enough_uring_writers_for_a_population_coexist_in_one_process`]
    /// went green at 160 writers in a `cargo test` binary that is on the system
    /// allocator and therefore never had the bug.
    ///
    /// **It measures the charge, not the count**, because the count is what the
    /// blind guard already measures. The charge is read off the kernel — the
    /// only place it exists — by binary-searching what can still be registered.
    ///
    /// **The positive control comes first and is not optional.** A box whose
    /// transparent huge pages are off cannot express this defect at all, and a
    /// guard that quietly passed there would be measuring nothing; so the huge
    /// page is built by hand, its charge is measured, and it must be at least
    /// 256 pages before the real assertion is believed.
    ///
    /// Noise: `RLIMIT_MEMLOCK` is charged uid-wide, so another process opening
    /// or closing a ring between the two readings moves a difference. Five
    /// readings are taken and the SMALLEST is asserted on — a false red would
    /// need five consecutive intrusions, and the two outcomes are 1 and 514, so
    /// there is no band where noise decides it.
    ///
    /// Seen RED by making `Staging::map` hand back a page out of a
    /// `MADV_HUGEPAGE` arena instead of its own mapping — which is exactly what
    /// `Box<[u8]>` did under mimalloc: *"writer 3 of 7 could not even be opened,
    /// with 428 page(s) of RLIMIT_MEMLOCK headroom left and the earlier writers
    /// charged [514, 512, 514] page(s) each. One page inside a transparent huge
    /// page costs 514 on this box…"*. Restored. Note the shape of that red: the
    /// bound below is not even reached, because four registrations is the whole
    /// budget — which is the server's measured ceiling of three repositories
    /// plus its control store, arrived at from the other end.
    #[test]
    fn a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in() {
        let _serial = measuring_memlock();
        let h = Headroom::new();

        // ── positive control ────────────────────────────────────────────────
        let Some(arena) = HugeArena::new() else {
            panic!(
                "this box would not give a transparent huge page (madvise refused), so it \
                 cannot exhibit the 512× registration charge this guard exists for. That is a \
                 property of the box, not a pass."
            );
        };
        let mut huge_charge = 0usize;
        for _ in 0..5 {
            let before = h.pages();
            let ctl = IoUring::new(RING_ENTRIES).expect("a control ring");
            let iov = libc::iovec {
                iov_base: arena.a_page() as *mut libc::c_void,
                iov_len: page_size(),
            };
            // SAFETY: `iov` names one page of the live arena above; the
            // registration is released before the arena is dropped.
            unsafe {
                ctl.submitter()
                    .register_buffers(std::slice::from_ref(&iov))
                    .expect("registering one page of a huge page")
            };
            let after = h.pages();
            huge_charge = huge_charge.max(before.saturating_sub(after));
            ctl.submitter()
                .unregister_buffers()
                .expect("release the control registration");
        }
        assert!(
            huge_charge >= 256,
            "the positive control charged only {huge_charge} page(s) for one page inside a \
             MADV_HUGEPAGE arena, so transparent huge pages are not actually in play here and \
             the assertion below would pass on any implementation"
        );

        // ── the writer itself ───────────────────────────────────────────────
        let dir = crate::store::tests::tmpdir("uring-registration-charge");
        let mut readings = Vec::new();
        for i in 0..7 {
            let before = h.pages();
            let w = UringWriter::create(&dir.join(format!("objects.pack.{i}")))
                .unwrap_or_else(|e| {
                    panic!(
                        "writer {i} of 7 could not even be opened, with {before} page(s) of \
                         RLIMIT_MEMLOCK headroom left and the earlier writers charged \
                         {readings:?} page(s) each. One page inside a transparent huge page \
                         costs {huge_charge} on this box, so a staging buffer that is not its \
                         own MADV_NOHUGEPAGE mapping caps a process at four registrations: \
                         {e:#}"
                    )
                });
            let after = h.pages();
            // Applied output, not just an open: the chain runs through the
            // registered buffer that was just measured.
            let (o, l) = w.append(b"one pack down the io_uring chain").unwrap();
            assert_eq!(
                read_journal(w.journal_file()).unwrap(),
                vec![(o, l)],
                "writer {i} acked without a journal row out of the registered buffer"
            );
            readings.push(before.saturating_sub(after));
            drop(w);
        }
        readings.sort_unstable();
        let charge = readings[readings.len() / 2];
        assert!(
            charge <= 64,
            "a UringWriter's registration was charged {charge} page(s) (readings {readings:?}). \
             One page inside a transparent huge page costs {huge_charge} on this box, and that \
             is what a `Box<[u8]>` buys you under an allocator that madvises its arenas — \
             mimalloc, which `gunnar serve` runs on. The staging buffer must be its own \
             MADV_NOHUGEPAGE mapping (`Staging::map`), which costs the ring plus exactly one \
             page."
        );
        // …and it costs SOMETHING. A charge of zero would mean the registration
        // this whole file is built on did not happen, and every bound above
        // would be satisfied by a writer that registers nothing at all.
        assert!(
            charge >= 1,
            "a UringWriter cost 0 page(s) of RLIMIT_MEMLOCK (readings {readings:?}), so nothing \
             was registered and the bound above is vacuous"
        );
        eprintln!(
            "load {}; a UringWriter costs {charge} page(s) of RLIMIT_MEMLOCK (readings \
             {readings:?}); one page inside a huge page costs {huge_charge}",
            loadavg(),
        );
    }

    /// **How many writers this box's `RLIMIT_MEMLOCK` has to be able to carry**,
    /// and the limit and page size it was derived from.
    ///
    /// Derived rather than hardcoded, so the two guards below state the same
    /// thing on any box: `n × 16 pages > limit`, so the 64 KiB registration this
    /// file carried until 2026-08-14 **cannot** pass by construction, and
    /// `n × 1 page ≤ limit / 4`, so a one-page registration passes with 4× of
    /// margin on the registration alone. The real margin is smaller and is
    /// stated where it is used: the ring costs ~5 more pages that this
    /// arithmetic cannot see, and the limit is charged uid-wide and shared with
    /// every other process this user is running.
    fn writers_the_limit_must_allow() -> (usize, usize, usize) {
        let page = page_size();
        let Some(limit) = memlock_limit_bytes().map(|b| b as usize) else {
            panic!(
                "RLIMIT_MEMLOCK is unlimited on this box, so it cannot exhibit the ceiling these \
                 guards exist for. That is a property of the box, not a pass: run them somewhere \
                 with the stock 8 MiB limit before believing the arm scales."
            );
        };
        // 16 pages is what this file registered per writer until 2026-08-14.
        let n = (limit / (16 * page) + 32).min(limit / (4 * page));
        assert!(
            n >= 8,
            "RLIMIT_MEMLOCK is only {limit} B on this box — too small for a guard to separate a \
             1-page registration from a 16-page one"
        );
        (n, limit, page)
    }

    /// **Enough io_uring stores to serve a population coexist in one process.**
    ///
    /// This is the guard for the failure the whole file's "what a registration
    /// costs" section is about. A `UringWriter` pins its registered staging
    /// buffer for as long as it is open, against a `RLIMIT_MEMLOCK` the kernel
    /// counts **per uid**; a server holds one writer per repository, so the
    /// number of repositories a process can serve on this arm is
    /// `RLIMIT_MEMLOCK / pinned-per-writer`. At 64 KiB per writer and the stock
    /// 8 MiB limit that is ~87, and gunnar's `multiuser_scaling` alone provisions
    /// 102 users with a repository each.
    ///
    /// `n` comes from [`writers_the_limit_must_allow`] — derived from this box's
    /// own limit rather than hardcoded.
    ///
    /// Every writer **appends** — the registration is not the assertion, the
    /// chain running through it is, and the last writer's journal is read back
    /// off disk to prove the row landed.
    ///
    /// Seen RED by restoring the old buffer size, `JOURNAL_STAGING = 64 * 1024`:
    /// "the io_uring arm ran out of pinned memory at writer 107 of 160:
    /// uring: register_buffers (65536 B, 16 page(s)): Cannot allocate memory
    /// (os error 12). …". Restored.
    ///
    /// **What it cannot see — and this one was expensive.** It counts writers,
    /// not what each of them is CHARGED, and the charge depends on which
    /// allocator the executable is on: a page out of a `madvise(MADV_HUGEPAGE)`d
    /// arena is charged 512. A `cargo test` binary is on the system allocator
    /// and never has that problem, so this guard was green at 160 writers on
    /// 2026-08-14 while `gunnar serve` — on mimalloc — could open **three**
    /// stores. [`a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in`]
    /// is the guard for that, and it measures the charge instead of counting.
    ///
    /// **What it cannot see.** `RLIMIT_MEMLOCK` is shared uid-wide, so another
    /// process of the same user holding pinned io_uring buffers eats the same
    /// budget. `n` is 160 on a stock 8 MiB limit against a measured ~400 that
    /// fit, so the margin is ~2.5× rather than the 4× the page arithmetic alone
    /// suggests — the ring itself costs ~5 pages that the arithmetic does not
    /// see. A box with an unlimited memlock cannot express this bug at all and
    /// the guard says so out loud rather than passing quietly.
    #[test]
    fn enough_uring_writers_for_a_population_coexist_in_one_process() {
        let _serial = measuring_memlock();
        let (n, limit, page) = writers_the_limit_must_allow();
        let dir = crate::store::tests::tmpdir("uring-memlock-ceiling");
        let payload = b"one pack down the io_uring chain".to_vec();
        let mut held: Vec<UringWriter> = Vec::with_capacity(n);
        let mut extents = Vec::with_capacity(n);
        for i in 0..n {
            let w = match UringWriter::create(&dir.join(format!("objects.pack.{i}"))) {
                Ok(w) => w,
                Err(e) => panic!(
                    "the io_uring arm ran out of pinned memory at writer {i} of {n}: {e:#}\n\
                     RLIMIT_MEMLOCK here is {limit} B and one page is {page} B, so {n} writers \
                     need {} B pinned. A server holds one writer per repository; this is the \
                     ceiling on how many repositories the arm can serve.",
                    n * page
                ),
            };
            extents.push(
                w.append(&payload)
                    .unwrap_or_else(|e| panic!("writer {i} of {n} could not append: {e:#}")),
            );
            held.push(w);
        }

        // Applied output, off disk, from the last writer standing: the pack
        // bytes verbatim at the extent it returned, and a journal row naming it.
        let last = held.last().unwrap();
        let (o, l) = *extents.last().unwrap();
        let mut back = vec![0u8; l as usize];
        last.blobs().read_exact_at(&mut back, o).unwrap();
        assert_eq!(back, payload, "writer {} did not store the bytes", n - 1);
        assert_eq!(
            read_journal(last.journal_file()).unwrap(),
            vec![(o, l)],
            "writer {} acked without a journal row naming the extent",
            n - 1
        );
        eprintln!(
            "load {}; {n} io_uring writers open at once, {} B pinned of a {limit} B \
             RLIMIT_MEMLOCK ({page} B/writer; the 64 KiB registration this replaced would have \
             needed {} B)",
            loadavg(),
            n * page,
            n * 16 * page,
        );
    }

    /// **A reopened io_uring writer APPENDS to its journal. It does not truncate
    /// it.**
    ///
    /// The journal is the durable half of §13.12's `indexed` bit and the only
    /// record that a pack was ever acked. `archive_write`'s module docs state
    /// for every durable arm that "a writer opened over an archive that already
    /// has a journal appends — it writes no second schema and it truncates
    /// nothing", and this arm did not obey it: `File::create` truncates, so a
    /// reopened store lost every earlier extent, re-queued nothing on the
    /// crash-recovery diff, and restarted pack ordinals at 0 — which is how a
    /// fresh pack takes the ordinal of one already absorbed and has its objects
    /// dropped (`indexer::packs_already_acked`).
    ///
    /// Asserted on the file: three appends across **three** writers over one
    /// archive, then the journal read back off disk with all three extents in
    /// append order. A single reopen would be enough for the truncation; the
    /// third proves the second writer did not simply start a second stream that
    /// `read_journal` stops at.
    ///
    /// Seen RED by restoring `File::create(&jpath)` (and the schema written
    /// unconditionally at offset 0): "the reopened io_uring writer lost the
    /// journal rows written before it: left: [(1024, 1024)] right: [(0, 512),
    /// (512, 512), (1024, 1024)]". Restored.
    #[test]
    fn a_reopened_uring_writer_appends_to_its_journal_rather_than_truncating_it() {
        let dir = crate::store::tests::tmpdir("uring-journal-reopen");
        let blobs = dir.join("objects.pack");
        let mut want = Vec::new();
        for (i, len) in [512usize, 512, 1024].into_iter().enumerate() {
            let w = UringWriter::create(&blobs)
                .unwrap_or_else(|e| panic!("open {i} of the same archive: {e:#}"));
            want.push(w.append(&vec![b'a' + i as u8; len]).unwrap());
            // Dropped here — the next iteration is a genuine reopen, including
            // the `flock` `open_blobs` takes.
        }
        assert_eq!(
            read_journal(&UringWriter::journal_path(&blobs)).unwrap(),
            want,
            "the reopened io_uring writer lost the journal rows written before it"
        );
        // …and the blob cursor resumed from the file rather than from zero, so
        // the extents do not overlap.
        assert_eq!(want, vec![(0, 512), (512, 512), (1024, 1024)]);
        eprintln!(
            "load {}; three io_uring writers over one archive: journal {:?}",
            loadavg(),
            want
        );
    }

    /// The encoded journal row really does fit the one page that is pinned for
    /// it — measured through the encoder both arms share, not assumed.
    ///
    /// It is asked of `(u64::MAX, u64::MAX)` rather than of a real extent: the
    /// row is fixed-width Arrow IPC, so the widest values are the honest
    /// question, and a guard that only ever encoded small offsets would be
    /// sitting on an identity value.
    ///
    /// Seen RED by `JOURNAL_STAGING = 128`: "a journal row is 224 B and the
    /// registered staging buffer is 128 B". Restored. That is the same refusal
    /// [`UringWriter::append`] raises at run time, which is why shrinking the
    /// registration is safe to do at all: a row that outgrew the page is named,
    /// never truncated.
    #[test]
    fn a_journal_row_fits_the_page_that_is_pinned_for_it() {
        let row = crate::archive_write::encode_journal_row(u64::MAX, u64::MAX).unwrap();
        assert!(
            row.len() <= JOURNAL_STAGING,
            "a journal row is {} B and the registered staging buffer is {JOURNAL_STAGING} B",
            row.len()
        );
        eprintln!(
            "load {}; journal row {} B into a {JOURNAL_STAGING} B registered buffer ({} page)",
            loadavg(),
            row.len(),
            JOURNAL_STAGING / page_size()
        );
    }
}