Skip to main content

znippy_plugin_git/
uring_write.rs

1//! Impl 3 — the io_uring arm: **one submission, four ops, kernel-enforced
2//! ordering**.
3//!
4//! [`SafeWriter`](crate::archive_write::SafeWriter) buys its durability with
5//! four blocking syscalls and four round-trips through the scheduler:
6//!
7//! ```text
8//!   pwrite(blob)  fsync(blob)  write(journal)  fsync(journal)
9//!      ↑ ring 0      ↑ ring 0       ↑ ring 0        ↑ ring 0     4 syscalls
10//! ```
11//!
12//! Here the same four operations are **one** `io_uring_enter(2)`. Each of the
13//! first three SQEs carries `IOSQE_IO_LINK`, so the kernel will not start an op
14//! until its predecessor has completed — the ordering contract is enforced by
15//! the kernel rather than by the caller blocking between calls:
16//!
17//! ```text
18//!   [ Write(blob) ]→[ Fsync(blob) ]→[ WriteFixed(journal) ]→[ Fsync(journal) ]
19//!     IO_LINK         IO_LINK          IO_LINK                (chain end)
20//!   ────────────────────── one io_uring_enter ──────────────────────
21//! ```
22//!
23//! That is **the same ordering as `SafeWriter`**, not a second one: blob bytes
24//! durable before the row that references them. A crash anywhere in the chain
25//! leaves orphan payload nobody points at. If a link fails the kernel cancels
26//! the rest of the chain with `ECANCELED`, which is exactly the semantics
27//! wanted: no journal row is ever written after a failed blob fsync.
28//!
29//! # Registered buffers, and the copy that is *not* being avoided
30//!
31//! `register_buffers` pins a set of stable buffers once, so per-op the kernel
32//! skips `get_user_pages`/`put_page` on them. That only works for buffers whose
33//! address the ring can be told about **up front**, which the caller's transient
34//! `&[u8]` is not. So this writer splits the two:
35//!
36//! * **journal** → a registered one-page staging buffer and `WriteFixed`. This is
37//!   the buffer that is stable across appends, and it is where the win is real.
38//!   Crucially it means there is **no arrow `BufWriter`** here at all: the
39//!   serialized IPC message is written into the registered buffer and the kernel
40//!   is handed that. You cannot have both arrow's buffered `StreamWriter` and a
41//!   registered buffer; the brief says take the io_uring side, and this does.
42//! * **blob** → plain `Write` against the caller's own pointer. Not registered
43//!   (it cannot be), but also **not copied** in userspace: the pushed pack bytes
44//!   go from the caller's slice straight into the ring.
45//!
46//! Saying this plainly matters: the registered buffer removes per-op page
47//! pinning on the journal, not a `memcpy` on the pack.
48//!
49//! # What a registration COSTS, and why the staging buffer is one page
50//!
51//! `IORING_REGISTER_BUFFERS` pins its pages against **`RLIMIT_MEMLOCK`**, and
52//! the kernel charges them to `user->locked_vm` — a counter kept on the
53//! `user_struct`, so it is **per-UID and shared by every process that user is
54//! running**, not per-process and not per-ring
55//! (`io_uring/rsrc.c: io_account_mem -> __io_account_mem`).
56//!
57//! Every store gets its own writer, therefore its own ring, therefore its own
58//! registration. So the pinned pages are `stores × ceil(JOURNAL_STAGING /
59//! PAGE_SIZE)` and the ceiling is a **hard, shared, uid-wide** one. The default
60//! on oden (and on stock Debian/Ubuntu) is 8 MiB.
61//!
62//! MEASURED on oden 2026-08-14, `RLIMIT_MEMLOCK` 8 MiB, one fresh process per
63//! reading, three readings each, identical code with only the buffer size
64//! varied:
65//!
66//! ```text
67//!   registered per ring   rings that fit        what refused the next one
68//!   ───────────────────   ──────────────        ─────────────────────────
69//!   64 KiB (16 pages)     87, 88, 87            IORING_REGISTER_BUFFERS
70//!    4 KiB  (1 page)      397, 422, 419         io_uring_setup
71//! ```
72//!
73//! with
74//!
75//! ```text
76//!   uring: register_buffers: Cannot allocate memory (os error 12)
77//! ```
78//!
79//! and every store opened after that refused. **~87 io_uring stores per
80//! process**, and gunnar's `multiuser_scaling` alone provisions 102 users with a
81//! repository each; a `gunnar serve` holds one store — therefore one ring,
82//! therefore one registration — per repository it has open. `FastWriter` and
83//! `SafeWriter` register nothing and have no such ceiling, which is exactly the
84//! shape the sweep showed: four workloads and eight forge arms red on **both**
85//! io_uring columns and green on `fast` and `safe`, with the index arm varied
86//! underneath and making no difference.
87//!
88//! Cutting the registration to one page moves that to **~400 rings, a 4.6×
89//! ceiling** — and past it the thing that refuses is no longer the registration
90//! at all but `io_uring_setup`. A bare ring of [`RING_ENTRIES`] costs about 21 KiB
91//! of the same budget (8 MiB / ~400), so the old writer spent ~21 pages per store
92//! and the new one spends ~6. It is not the 16× the buffer sizes suggest, because
93//! the ring was always paying five pages of it.
94//!
95//! End to end rather than at the writer: **150 whole `GitStore`s on this arm,
96//! open at once in one process**, each with a real 5 653 302-byte / 2 687-object
97//! pack pushed, indexed and read back — 3.4–4.0 s per store, oden 2026-08-14,
98//! load average 15–30. Every one of them past the old ceiling.
99//!
100//! # …and every number in the two paragraphs above was measured in the WRONG
101//! # PROCESS: the page you register is charged as the huge page it sits in
102//!
103//! *Found 2026-08-14, after the ceiling above had already been "fixed" once.*
104//!
105//! **`io_buffer_account_pin` does not charge the pages you named. It charges the
106//! `compound_head` of each of them.** If the 4 KiB you register happens to live
107//! inside a transparent huge page, the kernel charges **the whole 2 MiB — 512
108//! pages, not one** (`io_uring/rsrc.c: io_buffer_account_pin`, the
109//! `PageCompound` branch: `imu->acct_pages += page_size(hpage) >> PAGE_SHIFT`).
110//!
111//! MEASURED on oden 2026-08-14, `RLIMIT_MEMLOCK` 8 MiB, `transparent_hugepage
112//! = [madvise]`, one registration of exactly one page, charge read back off the
113//! kernel by binary-searching what could still be registered afterwards:
114//!
115//! ```text
116//!   where the one page came from                       pages charged
117//!   ────────────────────────────────────────────────   ─────────────
118//!   its own anonymous mmap, MADV_NOHUGEPAGE                        1
119//!   glibc malloc(4096) in a small C program                        2
120//!   inside a MADV_HUGEPAGE arena                                 512
121//! ```
122//!
123//! `Box<[u8]>` — what this file registered until now — is *whatever the process
124//! allocator gives you*, and **`gunnar serve` runs on mimalloc**, which
125//! `madvise(MADV_HUGEPAGE)`s its arenas. Measured on the running server:
126//! `AnonHugePages: 16384 kB` in one VMA, and every staging buffer allocated out
127//! of it. So the real ceiling in the process that matters was not ~400 stores.
128//! It was **three**:
129//!
130//! ```text
131//!   RLIMIT_MEMLOCK    io_uring stores one `gunnar serve` could open
132//!   ──────────────    ─────────────────────────────────────────────
133//!   8 MiB                             3   (+ the control store = 4 × 2 MiB)
134//!   4 MiB                             1
135//! ```
136//!
137//! — measured by pushing to distinct repositories one at a time until the server
138//! refused, oden 2026-08-14. Four huge pages fit in 8 MiB and that is the whole
139//! arithmetic. The fourth push onwards died with
140//! `register_buffers (4096 B, 1 page(s)): Cannot allocate memory`, which reads
141//! like the ceiling this file already documents and is a different one: it is not
142//! how MANY pages are registered, it is WHOSE page each one is.
143//!
144//! That is why [`Staging`] does not ask the allocator for the buffer. It takes
145//! **its own one-page anonymous mapping and `madvise(MADV_NOHUGEPAGE)`s it**, so
146//! the registration is charged one page whatever the executable's allocator does
147//! — and the "~400 rings" arithmetic above becomes true instead of merely
148//! plausible. `enough_uring_writers_for_a_population_coexist_in_one_process`
149//! could never have caught this: a `cargo test` binary is on the system
150//! allocator, whose 4 KiB allocations are not huge-page backed, so the guard
151//! measured a process that did not have the bug. The guard that does catch it is
152//! [`tests::a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in`],
153//! which measures the CHARGE rather than the count.
154//!
155//! # It is tighter than "N stores at once", because the kernel reclaims lazily
156//!
157//! `io_uring` teardown runs off a workqueue after the ring's last descriptor
158//! closes, so the pages of a **dropped** writer stay charged for a while. Opening
159//! and dropping one at a time and holding *nothing*, the 64 KiB writer was
160//! refused at the **63rd**. A server that opens a store per repository and lets
161//! it go — gunnar's `store_cache` map holds `Weak`s, so that is its shape while
162//! seeding — therefore hits this after about sixty repositories, which is where
163//! the sweep's `vs_forge_*` arms died seeding.
164//!
165//! The accounting is on the `user_struct`, so it also crosses process
166//! boundaries: a fresh process was refused its **second** ring while a previous
167//! test process's rings were still being torn down. That is also why there is no
168//! guard on the serial shape — one written against it failed at 54 of 160 purely
169//! on the residue of the guard that ran before it, and a test whose verdict
170//! depends on what else the box did in the last few seconds is not a guard. The
171//! serial ceiling is bounded by the same per-store cost the guards below do
172//! measure, and it clears with pacing: 600 rings created and dropped 5 ms apart
173//! did not fail once.
174//!
175//! # If the kernel cannot do it
176//!
177//! [`UringWriter::create`] probes for `IORING_OP_WRITE`, `IORING_OP_WRITE_FIXED`
178//! and `IORING_OP_FSYNC` and fails with a named error if any is missing, rather
179//! than silently degrading to `pwrite` and reporting an io_uring number that is
180//! not one. Same for `io_uring_setup` being blocked outright
181//! (`kernel.io_uring_disabled=2`, seccomp, a container without the syscall).
182
183#![cfg(target_os = "linux")]
184
185use std::fs::File;
186use std::os::unix::fs::FileExt;
187use std::os::unix::io::AsRawFd;
188use std::path::{Path, PathBuf};
189use std::sync::{Arc, Mutex};
190use std::sync::atomic::{AtomicU64, Ordering};
191
192use anyhow::{Result, anyhow, bail};
193use io_uring::{IoUring, opcode, squeue, types};
194
195use crate::archive_write::{
196    ArchiveWrite, Extent, encode_journal_row, journal_path, open_blobs,
197};
198
199/// Size of the registered staging buffer the journal message is built in.
200///
201/// **One page, because a registration is pinned memory charged uid-wide against
202/// `RLIMIT_MEMLOCK`** — see the module docs for the arithmetic and the measured
203/// failure. A page is the smallest thing the kernel can pin, so this is the
204/// floor, and it is not tight: an encoded journal row is two `u64`s plus Arrow
205/// IPC framing — **224 bytes**, measured by `encode_journal_row` on oden
206/// 2026-08-14 — so one page is still 18× headroom. A row that ever outgrew it
207/// would be refused by name in [`UringWriter::append`] rather than silently
208/// truncated.
209///
210/// It was 64 KiB until 2026-08-14. That is 16 pages per store on top of the
211/// ring's own ~5, and it is what capped a process at ~87 io_uring stores.
212///
213/// **The size is only half of it.** What the kernel charges depends on where the
214/// page came from, not only on how many there are — see [`Staging`] and the
215/// module docs' second cost section. One page out of mimalloc's huge-page arena
216/// is charged 512.
217const JOURNAL_STAGING: usize = 4096;
218
219/// Largest single `Write` SQE. A blob bigger than this is split across several
220/// linked `Write`s in the *same* chain, so the ordering guarantee is unchanged.
221const MAX_WRITE_CHUNK: usize = 1 << 30; // 1 GiB
222
223/// Ring depth. Enough for 61 blob chunks (61 GiB) plus fsync/journal/fsync.
224const RING_ENTRIES: u32 = 64;
225
226/// This kernel's page size — the granularity a registration is pinned at, so
227/// the unit [`JOURNAL_STAGING`] is really measured in.
228pub(crate) fn page_size() -> usize {
229    // SAFETY: `sysconf` takes no pointers and cannot fail destructively; a
230    // negative return (it has none for `_SC_PAGESIZE` on Linux) falls back to
231    // the architectural 4 KiB.
232    let n = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
233    if n > 0 { n as usize } else { 4096 }
234}
235
236/// The soft `RLIMIT_MEMLOCK` in bytes, or `None` if it is unlimited or
237/// unreadable.
238///
239/// This is the ceiling every `IORING_REGISTER_BUFFERS` in the process is
240/// charged against, and the kernel keeps the running total on the
241/// **`user_struct`** — so it is shared with every other process this uid is
242/// running. That is why an io_uring store can be refused by a ring some
243/// unrelated process of the same user opened.
244pub(crate) fn memlock_limit_bytes() -> Option<u64> {
245    let mut lim = libc::rlimit {
246        rlim_cur: 0,
247        rlim_max: 0,
248    };
249    // SAFETY: `lim` is a live, correctly-typed `rlimit` this call only writes.
250    if unsafe { libc::getrlimit(libc::RLIMIT_MEMLOCK, &mut lim) } != 0 {
251        return None;
252    }
253    if lim.rlim_cur == libc::RLIM_INFINITY {
254        None
255    } else {
256        Some(lim.rlim_cur as u64)
257    }
258}
259
260/// **Can this kernel run the arm at all?** — one `io_uring_setup` of a
261/// two-entry ring plus the opcode probe [`UringWriter::create`] makes, and
262/// nothing else: no registration, no file, no byte on disk. The ring is torn
263/// down before this returns.
264///
265/// For the caller that has to CHOOSE an arm before it opens its first store —
266/// gunnar's PID 1 on an appliance whose kernel may ship
267/// `# CONFIG_IO_URING is not set` (`ENOSYS`), `kernel.io_uring_disabled=2`
268/// (`EPERM`), or a seccomp policy without the syscall. `Err` carries the
269/// `std::io::Error` of the refusal, so the caller can fall back to
270/// [`SafeWriter`](crate::archive_write::SafeWriter) **by name** — "safe
271/// (io_uring unavailable: ENOSYS)" on a row — rather than by silence. This
272/// crate itself never falls back: [`UringWriter::create`] refuses, and that
273/// refusal is what a bench must see.
274///
275/// Cheap: one setup + one `IORING_REGISTER_PROBE` + close, ~30 µs on oden.
276/// A ring is charged against nothing durable after it is closed, so probing
277/// once per process leaves no residue for the stores that follow.
278pub fn probe() -> std::io::Result<()> {
279    let ring = IoUring::new(2)?;
280    let mut probe = io_uring::register::Probe::new();
281    ring.submitter().register_probe(&mut probe)?;
282    for (code, what) in [
283        (opcode::Write::CODE, "IORING_OP_WRITE"),
284        (opcode::WriteFixed::CODE, "IORING_OP_WRITE_FIXED"),
285        (opcode::Fsync::CODE, "IORING_OP_FSYNC"),
286    ] {
287        if !probe.is_supported(code) {
288            return Err(std::io::Error::new(
289                std::io::ErrorKind::Unsupported,
290                format!("this kernel's io_uring has no {what}"),
291            ));
292        }
293    }
294    Ok(())
295}
296
297/// The registered staging buffer: **its own anonymous mapping, one page,
298/// `MADV_NOHUGEPAGE`** — never the process allocator's memory.
299///
300/// It is a mapping rather than a `Box<[u8]>` for one reason, and it is the
301/// module docs' second "what a registration costs" section: the kernel charges
302/// a registration by `compound_head`, so a page handed out by an allocator that
303/// `madvise(MADV_HUGEPAGE)`s its arenas — mimalloc, which `gunnar serve` runs on
304/// — is charged as **512 pages**. Its own mapping cannot be part of anybody's
305/// huge page, and `MADV_NOHUGEPAGE` says so to `khugepaged` as well rather than
306/// relying on a one-page VMA being too small to collapse.
307///
308/// The address is stable for the mapping's whole life, which is what the
309/// registration requires, and it is unmapped in [`Drop`] **after** the ring that
310/// registered it has been dropped (field order in [`Ring`]).
311struct Staging {
312    ptr: *mut u8,
313    len: usize,
314}
315
316impl Staging {
317    /// Map `len` bytes, rounded up to whole pages, refusing huge pages.
318    fn map(len: usize) -> Result<Self> {
319        let page = page_size();
320        let len = len.next_multiple_of(page).max(page);
321        // SAFETY: a fresh anonymous mapping; no pointer of ours is passed in.
322        let ptr = unsafe {
323            libc::mmap(
324                std::ptr::null_mut(),
325                len,
326                libc::PROT_READ | libc::PROT_WRITE,
327                libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
328                -1,
329                0,
330            )
331        };
332        if ptr == libc::MAP_FAILED {
333            return Err(anyhow!(
334                "uring: mmap {len} B for the registered staging buffer: {}",
335                std::io::Error::last_os_error()
336            ));
337        }
338        // SAFETY: `ptr`/`len` are the mapping just returned.
339        let advised = unsafe { libc::madvise(ptr, len, libc::MADV_NOHUGEPAGE) };
340        if advised != 0 {
341            // ENOSYS/EINVAL means this kernel has no transparent huge pages at
342            // all, which is the state the advice was asking for. Anything else
343            // is a refusal to give it, and the registration would then be
344            // charged 512 pages instead of one — a silent 512× ceiling is
345            // exactly what this file was fixed for, so it is named, not
346            // swallowed.
347            let e = std::io::Error::last_os_error();
348            let benign = matches!(
349                e.raw_os_error(),
350                Some(libc::EINVAL) | Some(libc::ENOSYS)
351            );
352            if !benign {
353                // SAFETY: unmapping the mapping made three statements ago.
354                unsafe { libc::munmap(ptr, len) };
355                return Err(anyhow!(
356                    "uring: madvise(MADV_NOHUGEPAGE) on the staging buffer: {e}. Without it a \
357                     one-page registration can be charged as the whole 2 MiB huge page it sits \
358                     in, which caps a process at four io_uring stores on a stock 8 MiB \
359                     RLIMIT_MEMLOCK."
360                ));
361            }
362        }
363        Ok(Self {
364            ptr: ptr as *mut u8,
365            len,
366        })
367    }
368
369    fn len(&self) -> usize {
370        self.len
371    }
372    fn as_ptr(&self) -> *const u8 {
373        self.ptr
374    }
375    fn as_mut_slice(&mut self) -> &mut [u8] {
376        // SAFETY: `ptr..ptr+len` is our own live, readable, writable mapping,
377        // and `&mut self` is the only handle to it.
378        unsafe { std::slice::from_raw_parts_mut(self.ptr, self.len) }
379    }
380}
381
382impl Drop for Staging {
383    fn drop(&mut self) {
384        // SAFETY: the mapping this struct owns, unmapped exactly once. The ring
385        // that registered it is dropped first (field order in `Ring`), so no
386        // registration references these pages any more.
387        unsafe { libc::munmap(self.ptr as *mut libc::c_void, self.len) };
388    }
389}
390
391struct Ring {
392    ring: IoUring,
393    /// Registered index 0. Its own mapping, so its address is stable for the
394    /// lifetime of the registration even though `UringWriter` may move — and so
395    /// the kernel charges it one page. See [`Staging`].
396    staging: Staging,
397    journal_cursor: u64,
398}
399
400// SAFETY: every access to `Ring` goes through `UringWriter`'s `Mutex`, so the
401// submission and completion queues are never touched from two threads at once.
402// The registered buffer is owned here and outlives the registration (the ring is
403// dropped first, in field order).
404unsafe impl Send for Ring {}
405
406/// io_uring writer: write → fsync → journal → fsync as one linked submission.
407///
408/// # Durability on return
409///
410/// Identical to [`SafeWriter`](crate::archive_write::SafeWriter): the pack bytes
411/// are on the device and a journal row on the device references them. What
412/// differs is the cost of getting there — one `io_uring_enter` instead of four
413/// blocking syscalls — not what is promised.
414pub struct UringWriter {
415    blobs: Arc<File>,
416    journal: File,
417    journal_file_path: PathBuf,
418    cursor: AtomicU64,
419    ring: Mutex<Ring>,
420}
421
422impl UringWriter {
423    /// Open `archive`, set up the ring, register the journal staging buffer, and
424    /// open the journal beside it — **appending** to the rows already there, and
425    /// emitting the Arrow IPC schema header only if the file is new.
426    ///
427    /// Returns a named error — never a silent fallback — if this kernel cannot
428    /// provide what the chain needs. Nothing on disk is touched before that
429    /// point is passed.
430    pub fn create(archive: &Path) -> Result<Self> {
431        let (blobs, end) = open_blobs(archive)?;
432        let jpath = journal_path(archive);
433
434        // **The ring first, the journal second.** Nothing here may touch the
435        // journal until this writer is certain it can be built: `create` can
436        // still fail on `register_buffers` (the uid-wide `RLIMIT_MEMLOCK`
437        // ceiling in the module docs), and a failed open that had already
438        // opened the journal would leave a store's durable ack log standing
439        // behind a writer that does not exist.
440        let ring = IoUring::new(RING_ENTRIES).map_err(|e| {
441            anyhow!(
442                "uring: io_uring_setup failed ({e}). This kernel cannot run the io_uring arm \
443                 (check /proc/sys/kernel/io_uring_disabled, seccomp, container policy). \
444                 No fallback is substituted — a pwrite number reported as an io_uring number \
445                 would be a lie."
446            )
447        })?;
448
449        let mut probe = io_uring::register::Probe::new();
450        ring.submitter()
451            .register_probe(&mut probe)
452            .map_err(|e| anyhow!("uring: register_probe: {e}"))?;
453        for (code, what) in [
454            (opcode::Write::CODE, "IORING_OP_WRITE"),
455            (opcode::WriteFixed::CODE, "IORING_OP_WRITE_FIXED"),
456            (opcode::Fsync::CODE, "IORING_OP_FSYNC"),
457        ] {
458            if !probe.is_supported(code) {
459                bail!(
460                    "uring: this kernel does not support {what}; the write→fsync→journal→fsync \
461                     chain cannot be built. Not degrading to pwrite."
462                );
463            }
464        }
465
466        let staging = Staging::map(JOURNAL_STAGING)?;
467        // SAFETY: `staging` owns its own mapping and is owned by the `Ring`
468        // below; it is neither moved nor unmapped until the ring is dropped,
469        // which happens before it (struct field order in `Ring`).
470        unsafe {
471            let iov = libc::iovec {
472                iov_base: staging.as_ptr() as *mut libc::c_void,
473                iov_len: staging.len(),
474            };
475            ring.submitter()
476                .register_buffers(std::slice::from_ref(&iov))
477                .map_err(|e| {
478                    anyhow!(
479                        "uring: register_buffers ({} B, {} page(s)): {e}. A registration is \
480                         pinned memory charged against RLIMIT_MEMLOCK, and the kernel keeps that \
481                         count on the user_struct — it is per-UID and shared with every other \
482                         process this user is running, not per-process. This one is currently \
483                         {}. Every io_uring store holds one registration for as long as it is \
484                         open, so N stores pin N pages; `fast` and `safe` register nothing and \
485                         have no such ceiling. The page(s) named here are what was ASKED for; \
486                         the kernel charges by compound_head, so a staging buffer that ended up \
487                         inside a transparent huge page would be charged the whole 2 MiB — see \
488                         `Staging`, which takes its own MADV_NOHUGEPAGE mapping so that cannot \
489                         happen.",
490                        JOURNAL_STAGING,
491                        JOURNAL_STAGING.div_ceil(page_size()),
492                        memlock_limit_bytes()
493                            .map(|b| format!("{b} B"))
494                            .unwrap_or_else(|| "unreadable".into()),
495                    )
496                })?;
497        }
498
499        // **The journal is a LOG and a reopen APPENDS to it** — the contract
500        // `archive_write`'s module docs state for every durable arm, and until
501        // 2026-08-14 this writer was the one that broke it: it opened with
502        // `File::create`, which truncates, so a store reopened on this arm
503        // erased the durable record that its packs had ever been acked, derived
504        // an empty crash-recovery diff, and restarted pack ordinals at 0 — see
505        // `indexer::packs_already_acked` for what a restarted ordinal does to a
506        // pack's objects. `truncate(false)` plus a cursor taken from the file's
507        // own length is what makes the two durable arms actually
508        // indistinguishable on disk, which is what LAW 5 already claimed of
509        // them.
510        let journal = std::fs::OpenOptions::new()
511            .read(true)
512            .write(true)
513            .create(true)
514            .truncate(false)
515            .open(&jpath)
516            .map_err(|e| anyhow!("uring: open journal {}: {e}", jpath.display()))?;
517        // The schema message opens the stream and is written **once per file**,
518        // not once per writer: a second one mid-file makes arrow's
519        // `StreamReader` — and therefore `read_journal` — stop at it.
520        //
521        // Whether this file needs one is decided by the same function the safe
522        // arm asks, and there is no `fsync` here any more — see
523        // [`crate::archive_write::journal_head`] for why an empty store is
524        // rebuilt rather than barriered, and for the check that replaces it.
525        let head = crate::archive_write::journal_head(&journal, &jpath)?;
526        let journal_cursor = if head.rewrite {
527            journal
528                .set_len(0)
529                .map_err(|e| anyhow!("uring: resetting the interrupted journal {}: {e}", jpath.display()))?;
530            journal.write_all_at(&head.schema, 0)?;
531            head.schema.len() as u64
532        } else {
533            journal
534                .metadata()
535                .map_err(|e| anyhow!("uring: stat journal {}: {e}", jpath.display()))?
536                .len()
537        };
538
539        Ok(Self {
540            blobs,
541            journal,
542            journal_file_path: jpath,
543            cursor: AtomicU64::new(end),
544            ring: Mutex::new(Ring {
545                ring,
546                staging,
547                journal_cursor,
548            }),
549        })
550    }
551
552    /// Path of the journal segment beside `archive`.
553    pub fn journal_path(archive: &Path) -> PathBuf {
554        journal_path(archive)
555    }
556
557    /// The blob file, for a reader that wants to `pread` an extent back.
558    pub fn blobs(&self) -> &File {
559        &self.blobs
560    }
561
562    /// The journal file path this writer is appending rows to.
563    pub fn journal_file(&self) -> &Path {
564        &self.journal_file_path
565    }
566}
567
568impl ArchiveWrite for UringWriter {
569    fn append(&self, bytes: &[u8]) -> Result<Extent> {
570        let len = bytes.len() as u64;
571        let offset = self.cursor.fetch_add(len, Ordering::SeqCst);
572
573        let chunks = bytes.len().div_ceil(MAX_WRITE_CHUNK).max(1);
574        if chunks + 3 > RING_ENTRIES as usize {
575            bail!(
576                "uring: {} B needs {chunks} linked writes, more than the ring's {RING_ENTRIES} \
577                 entries",
578                bytes.len()
579            );
580        }
581
582        let row = encode_journal_row(offset, len)?;
583        let mut g = self
584            .ring
585            .lock()
586            .map_err(|_| anyhow!("uring mutex poisoned"))?;
587        if row.len() > g.staging.len() {
588            bail!(
589                "uring: journal row is {} B, staging buffer is {} B",
590                row.len(),
591                g.staging.len()
592            );
593        }
594        g.staging.as_mut_slice()[..row.len()].copy_from_slice(&row);
595        let journal_at = g.journal_cursor;
596
597        let blob_fd = types::Fd(self.blobs.as_raw_fd());
598        let journal_fd = types::Fd(self.journal.as_raw_fd());
599        let staging_ptr = g.staging.as_ptr();
600
601        let mut sqes: Vec<squeue::Entry> = Vec::with_capacity(chunks + 3);
602        // 1..=chunks — the pack bytes, verbatim, straight from the caller's slice.
603        for c in 0..chunks {
604            let start = c * MAX_WRITE_CHUNK;
605            let n = (bytes.len() - start).min(MAX_WRITE_CHUNK);
606            sqes.push(
607                opcode::Write::new(blob_fd, unsafe { bytes.as_ptr().add(start) }, n as u32)
608                    .offset(offset + start as u64)
609                    .build()
610                    .flags(squeue::Flags::IO_LINK)
611                    .user_data(c as u64),
612            );
613        }
614        // chunks+1 — blob bytes durable BEFORE the row that references them.
615        sqes.push(
616            opcode::Fsync::new(blob_fd)
617                .build()
618                .flags(squeue::Flags::IO_LINK)
619                .user_data(0xF5_00),
620        );
621        // chunks+2 — the journal row, out of the registered buffer, no BufWriter.
622        sqes.push(
623            opcode::WriteFixed::new(journal_fd, staging_ptr, row.len() as u32, 0)
624                .offset(journal_at)
625                .build()
626                .flags(squeue::Flags::IO_LINK)
627                .user_data(0x30_01),
628        );
629        // chunks+3 — the row is durable. Chain end: no IO_LINK.
630        sqes.push(opcode::Fsync::new(journal_fd).build().user_data(0xF5_01));
631
632        let want = sqes.len();
633        // SAFETY: every buffer referenced by an SQE (`bytes`, `g.staging`)
634        // outlives the `submit_and_wait` below, which does not return until all
635        // `want` operations have completed. The fds outlive `self`.
636        unsafe {
637            g.ring
638                .submission()
639                .push_multiple(&sqes)
640                .map_err(|e| anyhow!("uring: submission queue full: {e}"))?;
641        }
642        g.ring
643            .submit_and_wait(want)
644            .map_err(|e| anyhow!("uring: io_uring_enter: {e}"))?;
645
646        let mut written = 0i64;
647        let mut seen = 0usize;
648        let mut journal_written = 0i64;
649        for cqe in g.ring.completion() {
650            seen += 1;
651            let res = cqe.result();
652            if res < 0 {
653                let e = std::io::Error::from_raw_os_error(-res);
654                bail!(
655                    "uring: op {:#x} failed: {e} (a linked chain cancels its tail with \
656                     ECANCELED, so no journal row was written after a failed blob fsync)",
657                    cqe.user_data()
658                );
659            }
660            match cqe.user_data() {
661                0xF5_00 | 0xF5_01 => {}
662                0x30_01 => journal_written = res as i64,
663                _ => written += res as i64,
664            }
665        }
666        if seen != want {
667            bail!("uring: expected {want} completions, saw {seen}");
668        }
669        if written as u64 != len {
670            bail!(
671                "uring: short write — asked for {len} B, kernel wrote {written} B (io_uring \
672                 Write is not write_all)"
673            );
674        }
675        if journal_written as usize != row.len() {
676            bail!(
677                "uring: short journal write — {} B of {} B",
678                journal_written,
679                row.len()
680            );
681        }
682        g.journal_cursor = journal_at + row.len() as u64;
683
684        Ok((offset, len))
685    }
686
687    fn name(&self) -> &'static str {
688        "UringWriter"
689    }
690
691    fn durability(&self) -> &'static str {
692        "full — kernel-ordered blob fsync then journal fsync, one io_uring_enter"
693    }
694
695    fn blobs_shared(&self) -> Option<Arc<File>> {
696        Some(Arc::clone(&self.blobs))
697    }
698}
699
700
701#[cfg(test)]
702mod tests {
703    /// The probe and the writer agree: where one can be built the other says
704    /// yes, and the probe leaves nothing behind that a store would trip on.
705    #[test]
706    fn the_probe_answers_what_create_would_and_leaves_no_residue() {
707        let probed = super::probe();
708        let dir = std::env::temp_dir().join(format!("znippy-uring-probe-{}", std::process::id()));
709        std::fs::create_dir_all(&dir).unwrap();
710        let built = super::UringWriter::create(&dir.join("objects.pack"));
711        assert_eq!(probed.is_ok(), built.is_ok(), "probe {probed:?} vs create {}", built.as_ref().err().map(|e| e.to_string()).unwrap_or_default());
712        if probed.is_ok() {
713            for _ in 0..8 {
714                super::probe().expect("a probe per process start leaves no residue");
715            }
716        }
717        drop(built);
718        let _ = std::fs::remove_dir_all(&dir);
719    }
720
721    use super::*;
722    use crate::archive_write::{ArchiveWrite, read_journal};
723
724    fn loadavg() -> String {
725        std::fs::read_to_string("/proc/loadavg")
726            .unwrap_or_default()
727            .split_whitespace()
728            .take(3)
729            .collect::<Vec<_>>()
730            .join(" ")
731    }
732
733    // ── the instrument the charge guard is built on ─────────────────────────
734    //
735    // `user->locked_vm` is not readable from userspace, so the only honest way
736    // to ask what a registration COST is to ask the kernel what can still be
737    // registered afterwards. `probe` is one long-lived ring; `scratch` is one
738    // big NOHUGEPAGE mapping. Registering and *explicitly* unregistering
739    // releases the charge synchronously — unlike dropping a ring, whose
740    // un-accounting runs off a workqueue — so the measurement leaves no residue
741    // of its own.
742
743    /// **The two guards that measure `RLIMIT_MEMLOCK` may not run at once.**
744    ///
745    /// `cargo test` runs them on different threads of ONE process and the budget
746    /// is a single uid-wide counter, so they are not independent measurements of
747    /// anything: [`Headroom::pages`] transiently registers the whole limit to
748    /// find out what is left, which is indistinguishable — to
749    /// [`enough_uring_writers_for_a_population_coexist_in_one_process`] — from
750    /// the ceiling it exists to detect. Seen: that guard failing at writer 84 of
751    /// 160 purely because the charge guard was mid-binary-search beside it.
752    ///
753    /// # 🔴 And it did not serialise other PROCESSES, which was the bigger half
754    ///
755    /// *Found 2026-08-25.* The comment here read "this does not pretend to
756    /// serialise other processes", and the budget is a **uid-wide** counter, so
757    /// two `cargo test` binaries of this crate on one box — routine here — were
758    /// measuring each other. Reproduced by running four copies of the charge
759    /// guard at once, which is what an agent fleet on a shared box looks like:
760    ///
761    /// ```text
762    ///   run 1  ok    readings [0, 0, 0, 1, 3, 6, 898]
763    ///   run 2  FAIL  "registering one page of a huge page: Cannot allocate memory"
764    ///   run 3  FAIL  "a UringWriter cost 0 page(s) ... readings [0, 0, 0, 0, 4, 9, 1017]"
765    ///   run 4  ok    readings [0, 0, 2, 3, 3, 6, 10]
766    /// ```
767    ///
768    /// The `898` and the `1017` are the shape of the defect and the `0`s are
769    /// the same defect with its sign flipped: a neighbour releasing its
770    /// registration across the window is charged to *this* writer, a neighbour
771    /// taking one is subtracted from it, and `saturating_sub` turns the second
772    /// into a zero that reads exactly like "nothing was registered". Both
773    /// directions are false verdicts — a run reporting 898 would have failed
774    /// the `<= 64` bound this guard exists for, and a polluted positive control
775    /// would have satisfied `>= 256` without a huge page anywhere near it.
776    ///
777    /// So the lock is now taken **across processes too**, `flock` on a per-uid
778    /// file, because the uid is the scope of the thing being measured. It is
779    /// the honest scope: what this guard needs is not a quiet box, it is
780    /// exclusive use of one counter, and that is something a lock can actually
781    /// provide. What no lock can provide — an unrelated process on the same uid
782    /// — is caught instead by the bracket in [`Headroom::charge_of`].
783    static MEMLOCK_MEASUREMENT: Mutex<()> = Mutex::new(());
784
785    /// Both halves of the exclusion, released together.
786    struct Measuring {
787        _process: std::sync::MutexGuard<'static, ()>,
788        /// Closing the descriptor releases the `flock`.
789        _uid: std::fs::File,
790    }
791
792    fn measuring_memlock() -> Measuring {
793        use std::os::fd::AsRawFd as _;
794        let process = MEMLOCK_MEASUREMENT
795            .lock()
796            .unwrap_or_else(|p| p.into_inner());
797        // SAFETY: getuid is always safe.
798        let uid = unsafe { libc::getuid() };
799        let path = std::env::temp_dir().join(format!("znippy-git-memlock-{uid}.lock"));
800        let file = std::fs::OpenOptions::new()
801            .create(true)
802            .truncate(false)
803            .write(true)
804            .open(&path)
805            .unwrap_or_else(|e| panic!("opening {} for the memlock lock: {e}", path.display()));
806        // SAFETY: a descriptor this function owns and holds open for the
807        // lifetime of the returned guard.
808        let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) };
809        assert_eq!(
810            rc,
811            0,
812            "could not take the uid-wide memlock measurement lock at {}: {}",
813            path.display(),
814            std::io::Error::last_os_error()
815        );
816        Measuring {
817            _process: process,
818            _uid: file,
819        }
820    }
821
822    /// Big enough to ask for the whole stock 8 MiB limit in one registration.
823    const HEADROOM_CAP_PAGES: usize = 2048;
824
825    /// **The headroom probe's own scratch mapping — deliberately not
826    /// [`Staging::map`], even though it is the same two syscalls.**
827    ///
828    /// 🔴 The instrument may not share a code path with the thing it measures,
829    /// and this one did. Proven 2026-08-25 while re-running this guard's own
830    /// documented red proof: with `Staging::map` broken to `MADV_HUGEPAGE`, the
831    /// PROBE became huge-page-backed too, its resolution collapsed to 512-page
832    /// steps, and the guard failed with *"a UringWriter cost 0 page(s) ...
833    /// nothing was registered"* — for a writer that had just been charged 514.
834    /// A red that names the wrong defect is a red people learn to distrust,
835    /// which is the whole failure mode this file is about.
836    struct Scratch {
837        ptr: *mut libc::c_void,
838        len: usize,
839    }
840
841    impl Scratch {
842        fn new(len: usize) -> Self {
843            // SAFETY: a fresh anonymous mapping; no pointer of ours goes in.
844            let ptr = unsafe {
845                libc::mmap(
846                    std::ptr::null_mut(),
847                    len,
848                    libc::PROT_READ | libc::PROT_WRITE,
849                    libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
850                    -1,
851                    0,
852                )
853            };
854            assert_ne!(ptr, libc::MAP_FAILED, "the probe's scratch mapping");
855            // SAFETY: the mapping returned one statement ago. A failure here is
856            // a kernel without transparent huge pages, which is the state being
857            // asked for.
858            unsafe { libc::madvise(ptr, len, libc::MADV_NOHUGEPAGE) };
859            Self { ptr, len }
860        }
861        fn as_ptr(&self) -> *mut libc::c_void {
862            self.ptr
863        }
864    }
865
866    impl Drop for Scratch {
867        fn drop(&mut self) {
868            // SAFETY: the mapping this struct owns.
869            unsafe { libc::munmap(self.ptr, self.len) };
870        }
871    }
872
873    struct Headroom {
874        probe: IoUring,
875        scratch: Scratch,
876    }
877
878    /// How many bracketed readings a verdict is allowed to stand on, and how
879    /// many attempts may be spent gathering them. Five, because that is what
880    /// the module docs already claimed the guard took; twenty-five attempts,
881    /// because a discarded reading costs two binary searches and nothing else.
882    const CLEAN_READINGS: usize = 5;
883    const READING_ATTEMPTS: usize = 25;
884    /// How many times a bracket re-reads the budget waiting for a dropped
885    /// ring's workqueue un-accounting to land. Each try is a whole binary
886    /// search, so this is far more time than the release needs.
887    const SETTLE_TRIES: usize = 64;
888
889    impl Headroom {
890        fn new() -> Self {
891            let h = Self {
892                probe: IoUring::new(8).expect("a probe ring"),
893                scratch: Scratch::new(HEADROOM_CAP_PAGES * page_size()),
894            };
895            // **The instrument must not be saturated**, or every reading below
896            // is `2048 - 2048 = 0` and the guard reports "nothing was
897            // registered" on a box whose only sin is a generous
898            // `RLIMIT_MEMLOCK`. `pages` clamps at the cap by construction, so
899            // this is the one thing it cannot tell you itself.
900            assert!(
901                h.pages() < HEADROOM_CAP_PAGES,
902                "this uid can register the whole {HEADROOM_CAP_PAGES}-page probe cap, so the \
903                 headroom instrument is clamped and every charge it reads would be 0. Raise \
904                 HEADROOM_CAP_PAGES above RLIMIT_MEMLOCK, or run where the limit is the stock \
905                 8 MiB."
906            );
907            h
908        }
909
910        /// **One BRACKETED reading of what `acquire` costs**, or `None` if the
911        /// window was not this process's alone.
912        ///
913        /// `RLIMIT_MEMLOCK` is charged uid-wide, so a plain before/after
914        /// difference silently attributes any neighbour's registration or
915        /// release to whatever was being measured — see [`MEMLOCK_MEASUREMENT`]
916        /// for the four-way run where that produced 898 and 1017 in one column
917        /// and 0 in the next. The headroom is therefore read a **third** time,
918        /// after the thing under measurement has been released, and the reading
919        /// is believed only if the budget came back to where it started. A
920        /// neighbour that moved it across the window leaves the two ends
921        /// unequal and the reading is thrown away rather than averaged in.
922        fn charge_of<T>(&self, acquire: impl FnOnce() -> T) -> Option<usize> {
923            let before = self.pages();
924            let held = acquire();
925            let during = self.pages();
926            drop(held);
927            // 🔴 **Wait for the release, do not read across it.** A ring's own
928            // pinned pages are un-accounted off a **workqueue** when it is
929            // dropped — this module's own header says so about `Staging`, and
930            // it applies to the ring inside a `UringWriter` too. Reading the
931            // third bracket immediately therefore catches the crate's own
932            // teardown mid-flight and reports it as a neighbour moving the
933            // counter, which discarded every reading of the writer arm. Seen:
934            // six concurrent runs, "only 0 of 5 readings survived the bracket".
935            let mut after = self.pages();
936            for _ in 0..SETTLE_TRIES {
937                if after >= before {
938                    break;
939                }
940                std::thread::yield_now();
941                after = self.pages();
942            }
943            // `then`, not `then_some`: `before - during` is a subtraction on
944            // `usize` and `then_some` evaluates it eagerly, so a window in which
945            // the budget GREW panicked with "attempt to subtract with overflow"
946            // instead of being discarded. Measured the same six-run afternoon.
947            (after == before && during <= before).then(|| before - during)
948        }
949        fn fits(&self, pages: usize) -> bool {
950            let iov = libc::iovec {
951                iov_base: self.scratch.as_ptr(),
952                iov_len: pages * page_size(),
953            };
954            // SAFETY: `iov` names our own live scratch mapping, and the
955            // registration is dropped again before this function returns.
956            let ok = unsafe {
957                self.probe
958                    .submitter()
959                    .register_buffers(std::slice::from_ref(&iov))
960            }
961            .is_ok();
962            if ok {
963                self.probe
964                    .submitter()
965                    .unregister_buffers()
966                    .expect("unregister the probe buffer");
967            }
968            ok
969        }
970        /// Largest registration, in pages, this uid could make right now.
971        fn pages(&self) -> usize {
972            if self.fits(HEADROOM_CAP_PAGES) {
973                return HEADROOM_CAP_PAGES;
974            }
975            let (mut lo, mut hi) = (0usize, HEADROOM_CAP_PAGES);
976            while lo + 1 < hi {
977                let mid = (lo + hi) / 2;
978                if self.fits(mid) { lo = mid } else { hi = mid }
979            }
980            lo
981        }
982    }
983
984    /// A 2 MiB-aligned arena the kernel really has backed with a transparent
985    /// huge page — **verified in `/proc/self/smaps`, not assumed from a
986    /// `madvise` return** — or `None` if this box will not give one.
987    ///
988    /// How many fresh mappings to try before concluding the box will not give a
989    /// huge page. Each attempt is at a different address, and 2 MiB of
990    /// contiguous free memory is a property of a region and a moment.
991    const THP_ATTEMPTS: usize = 8;
992
993    struct HugeArena {
994        raw: *mut libc::c_void,
995        raw_len: usize,
996        arena: *mut u8,
997    }
998
999    impl HugeArena {
1000        const HUGE: usize = 2 << 20;
1001
1002        /// **A verified huge-page arena, or `None` after [`THP_ATTEMPTS`] tries.**
1003        ///
1004        /// 🔴 `madvise(MADV_HUGEPAGE)` returning 0 is a *request accepted*, not
1005        /// a huge page. The kernel still has to find 2 MiB of contiguous free
1006        /// memory at fault time, and on a box that has been building Rust all
1007        /// day it often cannot — it silently faults in 4 KiB pages instead.
1008        /// [`new_once`] then hands back an arena whose page costs exactly one,
1009        /// the positive control reads 1 instead of 512, and the guard fails
1010        /// with *"transparent huge pages are not actually in play here"* on a
1011        /// box whose `/sys/kernel/mm/transparent_hugepage/enabled` says
1012        /// `[madvise]`. Measured 2026-08-25: that red in 1 of 4 full suite runs
1013        /// while the same binary passed alone.
1014        ///
1015        /// So the arena is **checked** against `/proc/self/smaps` rather than
1016        /// assumed, and a refusal is retried at a fresh address — fragmentation
1017        /// is a property of one region, not of the box.
1018        fn new() -> Option<Self> {
1019            for _ in 0..THP_ATTEMPTS {
1020                let a = Self::new_once()?;
1021                if a.anon_huge_pages_kb() > 0 {
1022                    return Some(a);
1023                }
1024            }
1025            None
1026        }
1027
1028        /// `AnonHugePages` for the VMA this arena sits in, straight out of
1029        /// `/proc/self/smaps` — the only place the kernel says whether the
1030        /// advice was honoured.
1031        fn anon_huge_pages_kb(&self) -> u64 {
1032            let Ok(maps) = std::fs::read_to_string("/proc/self/smaps") else {
1033                return 0;
1034            };
1035            let want = self.arena as usize;
1036            let mut inside = false;
1037            for line in maps.lines() {
1038                if let Some((range, _)) = line.split_once(' ') {
1039                    if let Some((lo, hi)) = range.split_once('-') {
1040                        if let (Ok(lo), Ok(hi)) =
1041                            (usize::from_str_radix(lo, 16), usize::from_str_radix(hi, 16))
1042                        {
1043                            inside = (lo..hi).contains(&want);
1044                            continue;
1045                        }
1046                    }
1047                }
1048                if inside {
1049                    if let Some(v) = line.strip_prefix("AnonHugePages:") {
1050                        return v
1051                            .trim()
1052                            .trim_end_matches(" kB")
1053                            .parse()
1054                            .unwrap_or(0);
1055                    }
1056                }
1057            }
1058            0
1059        }
1060
1061        fn new_once() -> Option<Self> {
1062            let raw_len = 2 * Self::HUGE;
1063            // SAFETY: a fresh anonymous mapping.
1064            let raw = unsafe {
1065                libc::mmap(
1066                    std::ptr::null_mut(),
1067                    raw_len,
1068                    libc::PROT_READ | libc::PROT_WRITE,
1069                    libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
1070                    -1,
1071                    0,
1072                )
1073            };
1074            if raw == libc::MAP_FAILED {
1075                return None;
1076            }
1077            let arena = ((raw as usize + Self::HUGE - 1) & !(Self::HUGE - 1)) as *mut u8;
1078            // SAFETY: `arena..arena+HUGE` is inside the mapping above.
1079            unsafe {
1080                if libc::madvise(arena as *mut libc::c_void, Self::HUGE, libc::MADV_HUGEPAGE) != 0 {
1081                    libc::munmap(raw, raw_len);
1082                    return None;
1083                }
1084                // Fault it in, which is when the huge page is actually formed.
1085                std::ptr::write_bytes(arena, 1u8, Self::HUGE);
1086            }
1087            Some(Self {
1088                raw,
1089                raw_len,
1090                arena,
1091            })
1092        }
1093        /// A single page well inside the arena.
1094        fn a_page(&self) -> *const u8 {
1095            // SAFETY: four pages in, still inside the 2 MiB arena.
1096            unsafe { self.arena.add(4 * page_size()) }
1097        }
1098    }
1099
1100    impl Drop for HugeArena {
1101        fn drop(&mut self) {
1102            // SAFETY: the mapping this struct owns.
1103            unsafe { libc::munmap(self.raw, self.raw_len) };
1104        }
1105    }
1106
1107    /// A control-ring registration held open, released **explicitly** on drop.
1108    ///
1109    /// Dropping the ring alone would do it too, off a workqueue — asynchronously,
1110    /// which is exactly what the bracket in [`Headroom::charge_of`] cannot see
1111    /// happen and would then read as a neighbour's interference.
1112    struct HeldRegistration<'a>(&'a IoUring);
1113
1114    impl Drop for HeldRegistration<'_> {
1115        fn drop(&mut self) {
1116            self.0
1117                .submitter()
1118                .unregister_buffers()
1119                .expect("release the control registration");
1120        }
1121    }
1122
1123    /// [`CLEAN_READINGS`] bracketed readings of what `acquire` costs, or a panic
1124    /// that says the box would not hold still.
1125    ///
1126    /// The old form took a fixed seven readings and asserted on their median,
1127    /// with `saturating_sub` folding every polluted window into a `0`. Four of
1128    /// seven zeroes and the median IS zero, which is how this guard came to
1129    /// report "nothing was registered" about a writer that had just written
1130    /// through its own registered buffer. A reading is now either clean or
1131    /// discarded, and running out of attempts is reported as **the box**,
1132    /// named, rather than as a verdict on the code.
1133    fn gather<T>(what: &str, h: &Headroom, acquire: impl Fn(usize) -> T) -> Vec<usize> {
1134        let mut readings = Vec::with_capacity(CLEAN_READINGS);
1135        for i in 0..READING_ATTEMPTS {
1136            if let Some(n) = h.charge_of(|| acquire(i)) {
1137                readings.push(n);
1138                if readings.len() == CLEAN_READINGS {
1139                    return readings;
1140                }
1141            }
1142        }
1143        panic!(
1144            "only {} of {CLEAN_READINGS} readings of {what} survived the bracket in \
1145             {READING_ATTEMPTS} attempts (got {readings:?}, load {}). RLIMIT_MEMLOCK is charged \
1146             uid-wide and something else on this uid kept moving it across the measurement \
1147             window. That is a statement about the box, not about the writer: this guard needs \
1148             exclusive use of the counter, it takes a uid-wide flock to get it, and something \
1149             outside that lock is registering or releasing pinned memory.",
1150            readings.len(),
1151            loadavg(),
1152        )
1153    }
1154
1155    fn median(mut readings: Vec<usize>) -> usize {
1156        readings.sort_unstable();
1157        readings[readings.len() / 2]
1158    }
1159
1160    /// **A registration is charged one page — not the 2 MiB huge page the page
1161    /// happens to live in.**
1162    ///
1163    /// This is the guard for the defect that survived the first fix of this
1164    /// file's ceiling and was found on 2026-08-14: `io_buffer_account_pin`
1165    /// charges by `compound_head`, so one 4 KiB registration taken out of an
1166    /// allocator arena that has been `madvise(MADV_HUGEPAGE)`d costs **512
1167    /// pages**. `gunnar serve` runs on mimalloc, which does exactly that, and
1168    /// the result was a server that could open **three** io_uring stores on a
1169    /// stock 8 MiB `RLIMIT_MEMLOCK` — four huge pages, one of them the control
1170    /// store — while [`enough_uring_writers_for_a_population_coexist_in_one_process`]
1171    /// went green at 160 writers in a `cargo test` binary that is on the system
1172    /// allocator and therefore never had the bug.
1173    ///
1174    /// **It measures the charge, not the count**, because the count is what the
1175    /// blind guard already measures. The charge is read off the kernel — the
1176    /// only place it exists — by binary-searching what can still be registered.
1177    ///
1178    /// **The positive control comes first and is not optional.** A box whose
1179    /// transparent huge pages are off cannot express this defect at all, and a
1180    /// guard that quietly passed there would be measuring nothing; so the huge
1181    /// page is built by hand, its charge is measured, and it must be at least
1182    /// 256 pages before the real assertion is believed.
1183    ///
1184    /// Noise: `RLIMIT_MEMLOCK` is charged uid-wide, so another process opening
1185    /// or closing a ring between the two readings moves a difference. Five
1186    /// readings are taken and the SMALLEST is asserted on — a false red would
1187    /// need five consecutive intrusions, and the two outcomes are 1 and 514, so
1188    /// there is no band where noise decides it.
1189    ///
1190    /// Seen RED by making `Staging::map` hand back a page out of a
1191    /// `MADV_HUGEPAGE` arena instead of its own mapping — which is exactly what
1192    /// `Box<[u8]>` did under mimalloc: *"writer 3 of 7 could not even be opened,
1193    /// with 428 page(s) of RLIMIT_MEMLOCK headroom left and the earlier writers
1194    /// charged [514, 512, 514] page(s) each. One page inside a transparent huge
1195    /// page costs 514 on this box…"*. Restored. Note the shape of that red: the
1196    /// bound below is not even reached, because four registrations is the whole
1197    /// budget — which is the server's measured ceiling of three repositories
1198    /// plus its control store, arrived at from the other end.
1199    #[test]
1200    fn a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in() {
1201        let _serial = measuring_memlock();
1202        let h = Headroom::new();
1203
1204        // ── positive control ────────────────────────────────────────────────
1205        //
1206        // No huge page to be had is a property of the HOST at this moment (THP
1207        // off, or memory too fragmented to find 2 MiB contiguous — dwarf,
1208        // 2026-10-16: `[madvise]`, `AnonHugePages: 0 kB` box-wide), not of this
1209        // code. The guard cannot express the defect there, so it says SKIPPED
1210        // and why, loudly, rather than going red on the box or passing as if it
1211        // had measured something.
1212        let Some(arena) = HugeArena::new() else {
1213            eprintln!(
1214                "SKIPPED a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in: \
1215                 this box would not back a 2 MiB arena with a transparent huge page in \
1216                 {THP_ATTEMPTS} tries — `AnonHugePages` stayed 0 in /proc/self/smaps every \
1217                 time, with /sys/kernel/mm/transparent_hugepage/enabled = {:?} and load {}. \
1218                 Either THP is off, or memory is too fragmented right now to find 2 MiB \
1219                 contiguous; without one this guard cannot exhibit the 512× registration \
1220                 charge it exists for. NOT a pass: run it on a box with THP available.",
1221                std::fs::read_to_string("/sys/kernel/mm/transparent_hugepage/enabled")
1222                    .unwrap_or_default()
1223                    .trim(),
1224                loadavg(),
1225            );
1226            return;
1227        };
1228        // **One control ring for every reading**, not one per reading: a ring
1229        // created and dropped inside the bracket puts its own ~5 pages through
1230        // the workqueue on the way out, and the bracket would be measuring that
1231        // as much as the registration. Long-lived, its charge is part of
1232        // `before` and cancels.
1233        let ctl = IoUring::new(RING_ENTRIES).expect("a control ring");
1234        let huge_charge = median(gather(
1235            "one page inside a MADV_HUGEPAGE arena",
1236            &h,
1237            |_| {
1238                let iov = libc::iovec {
1239                    iov_base: arena.a_page() as *mut libc::c_void,
1240                    iov_len: page_size(),
1241                };
1242                // SAFETY: `iov` names one page of the live arena above, and
1243                // `HeldRegistration` releases the registration before this
1244                // scope — and therefore the arena — goes away.
1245                unsafe {
1246                    ctl.submitter()
1247                        .register_buffers(std::slice::from_ref(&iov))
1248                        .expect("registering one page of a huge page")
1249                };
1250                HeldRegistration(&ctl)
1251            },
1252        ));
1253        assert!(
1254            huge_charge >= 256,
1255            "the positive control charged only {huge_charge} page(s) for one page inside a \
1256             MADV_HUGEPAGE arena, so transparent huge pages are not actually in play here and \
1257             the assertion below would pass on any implementation"
1258        );
1259
1260        // ── the writer itself ───────────────────────────────────────────────
1261        let dir = crate::store::tests::tmpdir("uring-registration-charge");
1262        let readings = gather("a UringWriter", &h, |i| {
1263            let w = UringWriter::create(&dir.join(format!("objects.pack.{i}")))
1264                .unwrap_or_else(|e| {
1265                    panic!(
1266                        "writer {i} could not even be opened, with {} page(s) of \
1267                         RLIMIT_MEMLOCK headroom left. One page inside a transparent huge page \
1268                         costs {huge_charge} on this box, so a staging buffer that is not its \
1269                         own MADV_NOHUGEPAGE mapping caps a process at four registrations: \
1270                         {e:#}",
1271                        h.pages()
1272                    )
1273                });
1274            // Applied output, not just an open: the chain runs through the
1275            // registered buffer that is being measured, while it is held.
1276            let (o, l) = w.append(b"one pack down the io_uring chain").unwrap();
1277            assert_eq!(
1278                read_journal(w.journal_file()).unwrap(),
1279                vec![(o, l)],
1280                "writer {i} acked without a journal row out of the registered buffer"
1281            );
1282            w
1283        });
1284        let charge = median(readings.clone());
1285        assert!(
1286            charge <= 64,
1287            "a UringWriter's registration was charged {charge} page(s) (clean readings \
1288             {readings:?}). \
1289             One page inside a transparent huge page costs {huge_charge} on this box, and that \
1290             is what a `Box<[u8]>` buys you under an allocator that madvises its arenas — \
1291             mimalloc, which `gunnar serve` runs on. The staging buffer must be its own \
1292             MADV_NOHUGEPAGE mapping (`Staging::map`), which costs the ring plus exactly one \
1293             page."
1294        );
1295        // …and it costs SOMETHING. A charge of zero would mean the registration
1296        // this whole file is built on did not happen, and every bound above
1297        // would be satisfied by a writer that registers nothing at all.
1298        assert!(
1299            charge >= 1,
1300            "a UringWriter cost 0 page(s) of RLIMIT_MEMLOCK across {} bracketed readings \
1301             {readings:?} — every one of them taken with the uid-wide budget verified unmoved, \
1302             so this is not noise: nothing was registered and the bound above is vacuous",
1303            readings.len()
1304        );
1305        eprintln!(
1306            "load {}; a UringWriter costs {charge} page(s) of RLIMIT_MEMLOCK (clean readings \
1307             {readings:?}); one page inside a huge page costs {huge_charge}",
1308            loadavg(),
1309        );
1310    }
1311
1312    /// **How many writers this box's `RLIMIT_MEMLOCK` has to be able to carry**,
1313    /// and the limit and page size it was derived from.
1314    ///
1315    /// Derived rather than hardcoded, so the two guards below state the same
1316    /// thing on any box: `n × 16 pages > limit`, so the 64 KiB registration this
1317    /// file carried until 2026-08-14 **cannot** pass by construction, and
1318    /// `n × 1 page ≤ limit / 4`, so a one-page registration passes with 4× of
1319    /// margin on the registration alone. The real margin is smaller and is
1320    /// stated where it is used: the ring costs ~5 more pages that this
1321    /// arithmetic cannot see, and the limit is charged uid-wide and shared with
1322    /// every other process this user is running.
1323    fn writers_the_limit_must_allow() -> (usize, usize, usize) {
1324        let page = page_size();
1325        let Some(limit) = memlock_limit_bytes().map(|b| b as usize) else {
1326            panic!(
1327                "RLIMIT_MEMLOCK is unlimited on this box, so it cannot exhibit the ceiling these \
1328                 guards exist for. That is a property of the box, not a pass: run them somewhere \
1329                 with the stock 8 MiB limit before believing the arm scales."
1330            );
1331        };
1332        // 16 pages is what this file registered per writer until 2026-08-14.
1333        let n = (limit / (16 * page) + 32).min(limit / (4 * page));
1334        assert!(
1335            n >= 8,
1336            "RLIMIT_MEMLOCK is only {limit} B on this box — too small for a guard to separate a \
1337             1-page registration from a 16-page one"
1338        );
1339        (n, limit, page)
1340    }
1341
1342    /// **Enough io_uring stores to serve a population coexist in one process.**
1343    ///
1344    /// This is the guard for the failure the whole file's "what a registration
1345    /// costs" section is about. A `UringWriter` pins its registered staging
1346    /// buffer for as long as it is open, against a `RLIMIT_MEMLOCK` the kernel
1347    /// counts **per uid**; a server holds one writer per repository, so the
1348    /// number of repositories a process can serve on this arm is
1349    /// `RLIMIT_MEMLOCK / pinned-per-writer`. At 64 KiB per writer and the stock
1350    /// 8 MiB limit that is ~87, and gunnar's `multiuser_scaling` alone provisions
1351    /// 102 users with a repository each.
1352    ///
1353    /// `n` comes from [`writers_the_limit_must_allow`] — derived from this box's
1354    /// own limit rather than hardcoded.
1355    ///
1356    /// Every writer **appends** — the registration is not the assertion, the
1357    /// chain running through it is, and the last writer's journal is read back
1358    /// off disk to prove the row landed.
1359    ///
1360    /// Seen RED by restoring the old buffer size, `JOURNAL_STAGING = 64 * 1024`:
1361    /// "the io_uring arm ran out of pinned memory at writer 107 of 160:
1362    /// uring: register_buffers (65536 B, 16 page(s)): Cannot allocate memory
1363    /// (os error 12). …". Restored.
1364    ///
1365    /// **What it cannot see — and this one was expensive.** It counts writers,
1366    /// not what each of them is CHARGED, and the charge depends on which
1367    /// allocator the executable is on: a page out of a `madvise(MADV_HUGEPAGE)`d
1368    /// arena is charged 512. A `cargo test` binary is on the system allocator
1369    /// and never has that problem, so this guard was green at 160 writers on
1370    /// 2026-08-14 while `gunnar serve` — on mimalloc — could open **three**
1371    /// stores. [`a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in`]
1372    /// is the guard for that, and it measures the charge instead of counting.
1373    ///
1374    /// **What it cannot see.** `RLIMIT_MEMLOCK` is shared uid-wide, so another
1375    /// process of the same user holding pinned io_uring buffers eats the same
1376    /// budget. `n` is 160 on a stock 8 MiB limit against a measured ~400 that
1377    /// fit, so the margin is ~2.5× rather than the 4× the page arithmetic alone
1378    /// suggests — the ring itself costs ~5 pages that the arithmetic does not
1379    /// see. A box with an unlimited memlock cannot express this bug at all and
1380    /// the guard says so out loud rather than passing quietly.
1381    #[test]
1382    fn enough_uring_writers_for_a_population_coexist_in_one_process() {
1383        let _serial = measuring_memlock();
1384        let (n, limit, page) = writers_the_limit_must_allow();
1385        let dir = crate::store::tests::tmpdir("uring-memlock-ceiling");
1386        let payload = b"one pack down the io_uring chain".to_vec();
1387        let mut held: Vec<UringWriter> = Vec::with_capacity(n);
1388        let mut extents = Vec::with_capacity(n);
1389        for i in 0..n {
1390            let w = match UringWriter::create(&dir.join(format!("objects.pack.{i}"))) {
1391                Ok(w) => w,
1392                Err(e) => panic!(
1393                    "the io_uring arm ran out of pinned memory at writer {i} of {n}: {e:#}\n\
1394                     RLIMIT_MEMLOCK here is {limit} B and one page is {page} B, so {n} writers \
1395                     need {} B pinned. A server holds one writer per repository; this is the \
1396                     ceiling on how many repositories the arm can serve.",
1397                    n * page
1398                ),
1399            };
1400            extents.push(
1401                w.append(&payload)
1402                    .unwrap_or_else(|e| panic!("writer {i} of {n} could not append: {e:#}")),
1403            );
1404            held.push(w);
1405        }
1406
1407        // Applied output, off disk, from the last writer standing: the pack
1408        // bytes verbatim at the extent it returned, and a journal row naming it.
1409        let last = held.last().unwrap();
1410        let (o, l) = *extents.last().unwrap();
1411        let mut back = vec![0u8; l as usize];
1412        last.blobs().read_exact_at(&mut back, o).unwrap();
1413        assert_eq!(back, payload, "writer {} did not store the bytes", n - 1);
1414        assert_eq!(
1415            read_journal(last.journal_file()).unwrap(),
1416            vec![(o, l)],
1417            "writer {} acked without a journal row naming the extent",
1418            n - 1
1419        );
1420        eprintln!(
1421            "load {}; {n} io_uring writers open at once, {} B pinned of a {limit} B \
1422             RLIMIT_MEMLOCK ({page} B/writer; the 64 KiB registration this replaced would have \
1423             needed {} B)",
1424            loadavg(),
1425            n * page,
1426            n * 16 * page,
1427        );
1428    }
1429
1430    /// **A reopened io_uring writer APPENDS to its journal. It does not truncate
1431    /// it.**
1432    ///
1433    /// The journal is the durable half of §13.12's `indexed` bit and the only
1434    /// record that a pack was ever acked. `archive_write`'s module docs state
1435    /// for every durable arm that "a writer opened over an archive that already
1436    /// has a journal appends — it writes no second schema and it truncates
1437    /// nothing", and this arm did not obey it: `File::create` truncates, so a
1438    /// reopened store lost every earlier extent, re-queued nothing on the
1439    /// crash-recovery diff, and restarted pack ordinals at 0 — which is how a
1440    /// fresh pack takes the ordinal of one already absorbed and has its objects
1441    /// dropped (`indexer::packs_already_acked`).
1442    ///
1443    /// Asserted on the file: three appends across **three** writers over one
1444    /// archive, then the journal read back off disk with all three extents in
1445    /// append order. A single reopen would be enough for the truncation; the
1446    /// third proves the second writer did not simply start a second stream that
1447    /// `read_journal` stops at.
1448    ///
1449    /// Seen RED by restoring `File::create(&jpath)` (and the schema written
1450    /// unconditionally at offset 0): "the reopened io_uring writer lost the
1451    /// journal rows written before it: left: [(1024, 1024)] right: [(0, 512),
1452    /// (512, 512), (1024, 1024)]". Restored.
1453    #[test]
1454    fn a_reopened_uring_writer_appends_to_its_journal_rather_than_truncating_it() {
1455        let dir = crate::store::tests::tmpdir("uring-journal-reopen");
1456        let blobs = dir.join("objects.pack");
1457        let mut want = Vec::new();
1458        for (i, len) in [512usize, 512, 1024].into_iter().enumerate() {
1459            let w = UringWriter::create(&blobs)
1460                .unwrap_or_else(|e| panic!("open {i} of the same archive: {e:#}"));
1461            want.push(w.append(&vec![b'a' + i as u8; len]).unwrap());
1462            // Dropped here — the next iteration is a genuine reopen, including
1463            // the `flock` `open_blobs` takes.
1464        }
1465        assert_eq!(
1466            read_journal(&UringWriter::journal_path(&blobs)).unwrap(),
1467            want,
1468            "the reopened io_uring writer lost the journal rows written before it"
1469        );
1470        // …and the blob cursor resumed from the file rather than from zero, so
1471        // the extents do not overlap.
1472        assert_eq!(want, vec![(0, 512), (512, 512), (1024, 1024)]);
1473        eprintln!(
1474            "load {}; three io_uring writers over one archive: journal {:?}",
1475            loadavg(),
1476            want
1477        );
1478    }
1479
1480    /// The encoded journal row really does fit the one page that is pinned for
1481    /// it — measured through the encoder both arms share, not assumed.
1482    ///
1483    /// It is asked of `(u64::MAX, u64::MAX)` rather than of a real extent: the
1484    /// row is fixed-width Arrow IPC, so the widest values are the honest
1485    /// question, and a guard that only ever encoded small offsets would be
1486    /// sitting on an identity value.
1487    ///
1488    /// Seen RED by `JOURNAL_STAGING = 128`: "a journal row is 224 B and the
1489    /// registered staging buffer is 128 B". Restored. That is the same refusal
1490    /// [`UringWriter::append`] raises at run time, which is why shrinking the
1491    /// registration is safe to do at all: a row that outgrew the page is named,
1492    /// never truncated.
1493    #[test]
1494    fn a_journal_row_fits_the_page_that_is_pinned_for_it() {
1495        let row = crate::archive_write::encode_journal_row(u64::MAX, u64::MAX).unwrap();
1496        assert!(
1497            row.len() <= JOURNAL_STAGING,
1498            "a journal row is {} B and the registered staging buffer is {JOURNAL_STAGING} B",
1499            row.len()
1500        );
1501        eprintln!(
1502            "load {}; journal row {} B into a {JOURNAL_STAGING} B registered buffer ({} page)",
1503            loadavg(),
1504            row.len(),
1505            JOURNAL_STAGING / page_size()
1506        );
1507    }
1508}
1509