air_sys_syscall/io_uring/

provided.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! **Buffers fournis ring-mapped** (Temps 3b, `IORING_REGISTER_PBUF_RING`) :
//! l'application enregistre un **groupe de buffers** ([`ProvidedBufferRing`]) ;
//! elle soumet un `recv`/`read` avec `IOSQE_BUFFER_SELECT` **sans** buffer
//! attaché ; le kernel **choisit un buffer du groupe** au moment où les données
//! arrivent et rend son **id** dans le CQE (`CQE_F_BUFFER`, 16 bits hauts de
//! `cqe->flags`). Les buffers ne sont consommés que par les connexions actives.
//!
//! Référence normative : `docs/specs/layer-0/io-uring-3b-provided.md`.
//!
//! **Mécanisme.** Un groupe possède deux mémoires : l'**anneau de descripteurs**
//! (`io_uring_buf[count]`, partagé avec le kernel — le `tail` `u16` à l'offset 14
//! du premier descripteur est avancé par **store-release**) et la **mémoire de
//! données** (`count × buf_size`, possédée). L'anneau de descripteurs est soit
//! alloué par la façade (mmap anonyme page-aligné), soit par le kernel
//! (`kernel_mmap`) puis mmappé à `IORING_OFF_PBUF_RING | (group_id << SHIFT)`.
//!
//! **Cycle checkout → traitement → return** rendu sûr par RAII
//! ([`ProvidedBuffer`]) : la libération du guard **réapprovisionne** le groupe
//! (réécrit le descripteur, avance le `tail`). En mode **incrémental**
//! (`IOU_PBUF_RING_INC`), un `CQE_F_BUF_MORE` signale que **le même buffer**
//! recevra d'autres complétions → le guard **ne réapprovisionne pas** (réappro
//! seulement à la complétion finale, sans `BUF_MORE`).
//!
//! **Legacy évacués** : `PROVIDE_BUFFERS` (31) / `REMOVE_BUFFERS` (32) — cf.
//! `docs/UNSUPPORTED.md`.

use super::{Completion, IoUring, SubmissionToken, raw, syscall};
use crate::mem::{Mapping, MmapRegion, mmap_file};
use air_sys_types::Errno;
use air_sys_types::fd::{AsRawFd, BorrowedFd};
use air_sys_types::io_uring::CompletionFlags;
use air_sys_types::mem::{MapFlags, ProtectionFlags};
use air_sys_types::net::MessageFlags;
use alloc::vec::Vec;
use core::num::{NonZeroU16, NonZeroU32};
use core::sync::atomic::{AtomicU16, Ordering};

// ───────────────────────────────────────────────────────────────────────────
// Options & status
// ───────────────────────────────────────────────────────────────────────────

/// Options de création d'un [`ProvidedBufferRing`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ProvidedBufferRingOptions {
    /// Mémoire de l'anneau de **descripteurs** allouée par le **kernel**
    /// (`IOU_PBUF_RING_MMAP`) puis mmappée par la façade, au lieu d'être fournie
    /// par l'application.
    pub kernel_mmap: bool,
    /// Consommation **incrémentale** (`IOU_PBUF_RING_INC`) : un buffer peut servir
    /// plusieurs complétions, consommé au fur et à mesure (cf. module §5).
    pub incremental: bool,
}

/// État d'un groupe rendu par [`ProvidedBufferRing::status`] (`PBUF_STATUS`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ProvidedBufferRingStatus {
    /// Tête courante du groupe (index de consommation côté kernel).
    pub head: u32,
}

// ───────────────────────────────────────────────────────────────────────────
// ProvidedBufferRing
// ───────────────────────────────────────────────────────────────────────────

/// Origine de la mémoire de l'**anneau de descripteurs** (page-alignée).
#[derive(Debug)]
enum RingMemory {
    /// Anneau alloué par la façade (mmap anonyme).
    App(MmapRegion),
    /// Anneau alloué par le kernel (`IOU_PBUF_RING_MMAP`) puis mmappé.
    Kernel(Mapping),
}

impl RingMemory {
    /// Base de l'anneau (`struct io_uring_buf_ring`). `*mut` : la façade écrit
    /// les descripteurs et le `tail` ; le kernel lit les descripteurs/`tail` et
    /// écrit le `head` (champs **disjoints**, pas d'écriture aliasée).
    fn base(&self) -> *mut u8 {
        match self {
            Self::App(region) => region.as_ptr().cast_mut(),
            Self::Kernel(mapping) => mapping.as_ptr().cast_mut(),
        }
    }
}

/// Groupe de **buffers fournis** *ring-mapped* (`IORING_REGISTER_PBUF_RING`, 22).
///
/// Possède l'anneau de descripteurs **et** la mémoire de données tant que le
/// groupe est enregistré ; [`ProvidedBufferRing::unregister`] les restitue
/// (désenregistre puis libère). Les buffers consommés sont **rendus** au groupe
/// par la libération du guard [`ProvidedBuffer`].
#[derive(Debug)]
pub struct ProvidedBufferRing {
    /// Identifiant de groupe (`bgid`).
    bgid: u16,
    /// Nombre de buffers (puissance de 2).
    count: u16,
    /// `count - 1` (masque d'indexation de l'anneau).
    mask: u16,
    /// Taille d'un buffer en octets.
    buf_size: u32,
    /// Mémoire de l'anneau de descripteurs (partagée avec le kernel).
    ring_mem: RingMemory,
    /// Mémoire de données possédée (`count × buf_size` octets). Le kernel y écrit
    /// les données reçues ; reste vivante tant que le groupe est enregistré.
    data: Vec<u8>,
    /// `tail` local (prochain slot où écrire un descripteur ; wrappe).
    provide_tail: u16,
    /// Nombre de buffers actuellement disponibles côté groupe (non checkout).
    available: u16,
    /// Buffer `bid` actuellement **sorti** (checkout non encore rendu). Indexé par
    /// `bid` ; rend l'accounting d'`available` **idempotent** vis-à-vis des
    /// complétions multiples d'un même buffer (mode incrémental, `BUF_MORE`).
    out_flags: Vec<bool>,
}

impl ProvidedBufferRing {
    /// `IORING_REGISTER_PBUF_RING` (22) : enregistre un groupe `group_id` de
    /// `count` buffers de `buf_size` octets (`count` **puissance de 2**), puis
    /// approvisionne les `count` buffers.
    ///
    /// # Errors
    ///
    /// [`Errno::EINVAL`] si `count` n'est pas une puissance de 2 ou si
    /// `count × buf_size` déborde ; erreurs de `mmap`/`io_uring_register`
    /// ([`Errno::ENOMEM`], [`Errno::EEXIST`] si `group_id` déjà pris).
    pub fn register(
        ring: &mut IoUring,
        group_id: u16,
        count: NonZeroU16,
        buf_size: NonZeroU32,
        opts: ProvidedBufferRingOptions,
    ) -> Result<Self, Errno> {
        let count = count.get();
        if !count.is_power_of_two() {
            return Err(Errno::EINVAL);
        }
        let buf_size = buf_size.get();
        let count_usize = usize::from(count);
        let total = count_usize
            .checked_mul(usize::try_from(buf_size).expect("u32 ⊆ usize sur cible LP64"))
            .ok_or(Errno::EINVAL)?;
        let ring_bytes = count_usize
            .checked_mul(raw::IO_URING_BUF_SIZE)
            .ok_or(Errno::EINVAL)?;
        let data = vec![0u8; total];

        let mut flags = if opts.incremental {
            raw::IOU_PBUF_RING_INC
        } else {
            0
        };
        // Mode app : anneau alloué ici (mmap anonyme page-aligné).
        let app_region = if opts.kernel_mmap {
            flags |= raw::IOU_PBUF_RING_MMAP;
            None
        } else {
            Some(MmapRegion::new_anonymous(
                ring_bytes,
                ProtectionFlags::READ | ProtectionFlags::WRITE,
                MapFlags::PRIVATE,
            )?)
        };
        let ring_addr = app_region.as_ref().map_or(0, |r| r.as_ptr() as u64);

        let reg = raw::IoUringBufReg {
            ring_addr,
            ring_entries: u32::from(count),
            bgid: group_id,
            flags,
            resv: [0; 3],
        };
        // SAFETY: PBUF_RING attend une `io_uring_buf_reg` (nr_args == 1) ; `reg`
        // vit pendant l'appel ; `ring_addr` pointe l'anneau page-aligné (mode app)
        // ou 0 (mode kernel).
        let ret = unsafe {
            syscall::register(
                ring.fd_raw(),
                raw::IORING_REGISTER_PBUF_RING,
                core::ptr::from_ref(&reg) as u64,
                1,
            )
        };
        if ret < 0 {
            return Err(raw::errno_from_negative_syscall_ret(ret));
        }

        // Mode kernel : mmap l'anneau alloué par le kernel.
        let ring_mem = match app_region {
            Some(region) => RingMemory::App(region),
            None => {
                let offset =
                    raw::IORING_OFF_PBUF_RING | (u64::from(group_id) << raw::IORING_OFF_PBUF_SHIFT);
                // SAFETY: `ring.fd_raw()` est un ring fd valide possédé, emprunté
                // le temps de l'appel mmap.
                let fd = unsafe { BorrowedFd::borrow_raw(ring.fd_raw()) };
                let mapping = mmap_file(
                    fd,
                    ring_bytes,
                    offset,
                    ProtectionFlags::READ | ProtectionFlags::WRITE,
                    MapFlags::SHARED,
                )?;
                RingMemory::Kernel(mapping)
            }
        };

        let mut this = Self {
            bgid: group_id,
            count,
            mask: count.wrapping_sub(1),
            buf_size,
            ring_mem,
            data,
            provide_tail: 0,
            available: 0,
            out_flags: vec![false; usize::from(count)],
        };
        this.provide_initial();
        Ok(this)
    }

    /// Approvisionne les `count` buffers à l'enregistrement (descripteurs
    /// `0..count`, `tail = count`).
    fn provide_initial(&mut self) {
        for bid in 0..self.count {
            // Slot == bid à l'initialisation (tail part de 0).
            self.write_descriptor(bid, bid);
        }
        self.provide_tail = self.count;
        self.available = self.count;
        self.publish_tail();
    }

    /// Écrit un descripteur `io_uring_buf` (`addr`/`len`/`bid`) dans `slot`, sans
    /// toucher l'offset 14 (`resv`/`tail`).
    fn write_descriptor(&self, slot: u16, bid: u16) {
        let base = self.ring_mem.base();
        let slot_off = usize::from(slot & self.mask)
            .checked_mul(raw::IO_URING_BUF_SIZE)
            .expect("slot borné par le masque");
        let data_off = usize::from(bid)
            .checked_mul(usize::try_from(self.buf_size).expect("u32 ⊆ usize"))
            .expect("bid × buf_size ≤ total (borné par construction)");
        let addr = self.data.as_ptr().wrapping_add(data_off) as u64;
        // SAFETY: `base` pointe l'anneau page-aligné vivant ; `slot_off` est dans
        // `count × 16` octets (slot masqué) ; les écritures `u64`/`u32`/`u16` sont
        // alignées (base page-alignée, offsets multiples de 8/4/2). On n'écrit pas
        // l'offset 14 (`tail`). `addr` désigne le buffer `bid` dans `data` (vivant).
        unsafe {
            let buf = base.add(slot_off);
            buf.cast::<u64>().write(addr);
            buf.add(8).cast::<u32>().write(self.buf_size);
            buf.add(12).cast::<u16>().write(bid);
        }
    }

    /// Publie le `tail` local par un **store-release** (rend les descripteurs
    /// écrits visibles au kernel avant l'avancée de la queue).
    fn publish_tail(&self) {
        let base = self.ring_mem.base();
        // SAFETY: `base + 14` est le mot `tail` (`u16`, 2-aligné) de l'anneau
        // vivant ; `AtomicU16` a le même layout qu'un `u16`. Store-release apparié
        // au load-acquire du kernel (protocole d'anneau de buffers).
        unsafe {
            let tail = base
                .add(raw::IO_URING_BUF_RING_TAIL_OFFSET)
                .cast::<AtomicU16>();
            (*tail).store(self.provide_tail, Ordering::Release);
        }
    }

    /// Comptabilise un **checkout** du buffer `bid` (le kernel l'a sorti pour une
    /// complétion). Idempotent : une complétion supplémentaire sur un buffer
    /// **déjà** sorti (incrémental, `BUF_MORE`) ne re-décrémente pas `available`.
    fn checkout(&mut self, bid: u16) {
        let newly_out = self
            .out_flags
            .get_mut(usize::from(bid))
            .is_some_and(|flag| {
                if *flag {
                    false
                } else {
                    *flag = true;
                    true
                }
            });
        if newly_out {
            self.available = self.available.saturating_sub(1);
        }
    }

    /// **Rend** le buffer `bid` au groupe : réécrit son descripteur au `tail`,
    /// avance et publie le `tail`, et — s'il était sorti — incrémente `available`.
    fn return_buffer(&mut self, bid: u16) {
        self.write_descriptor(self.provide_tail, bid);
        self.provide_tail = self.provide_tail.wrapping_add(1);
        self.publish_tail();
        let was_out = self
            .out_flags
            .get_mut(usize::from(bid))
            .is_some_and(|flag| {
                let prev = *flag;
                *flag = false;
                prev
            });
        if was_out {
            self.available = self.available.saturating_add(1).min(self.count);
        }
    }

    /// Tranche d'octets du buffer `bid`, longueur `len` (bornée par construction :
    /// `len ≤ buf_size` et `bid < count`).
    fn buffer_bytes(&self, bid: u16, len: usize) -> &[u8] {
        let bufsz = usize::try_from(self.buf_size).expect("u32 ⊆ usize");
        let start = usize::from(bid)
            .checked_mul(bufsz)
            .expect("bid × buf_size borné");
        let end = start.checked_add(len).expect("start + len borné");
        self.data
            .get(start..end)
            .expect("buffer borné (len ≤ buf_size, bid < count)")
    }

    /// Variante mutable de [`Self::buffer_bytes`].
    fn buffer_bytes_mut(&mut self, bid: u16, len: usize) -> &mut [u8] {
        let bufsz = usize::try_from(self.buf_size).expect("u32 ⊆ usize");
        let start = usize::from(bid)
            .checked_mul(bufsz)
            .expect("bid × buf_size borné");
        let end = start.checked_add(len).expect("start + len borné");
        self.data
            .get_mut(start..end)
            .expect("buffer borné (len ≤ buf_size, bid < count)")
    }

    /// `IORING_UNREGISTER_PBUF_RING` (23) : désenregistre le groupe et **libère**
    /// l'anneau et la mémoire de données (à la consommation de `self`).
    ///
    /// **Sûreté par construction** : un [`ProvidedBuffer`] en cours de traitement
    /// emprunte le groupe `&mut` — impossible de désenregistrer tant qu'il vit.
    /// Le code suivant **ne compile pas** :
    ///
    /// ```compile_fail
    /// # use air_sys_syscall::io_uring::{IoUring, ProvidedBufferRing, ProvidedBufferRingOptions};
    /// # use core::num::{NonZeroU16, NonZeroU32};
    /// # fn demo(ring: &mut IoUring, completion: air_sys_syscall::io_uring::Completion) {
    /// let mut group = ProvidedBufferRing::register(
    ///     ring, 0, NonZeroU16::new(4).unwrap(), NonZeroU32::new(64).unwrap(),
    ///     ProvidedBufferRingOptions::default()).unwrap();
    /// let buf = completion.into_provided_buffer(&mut group).unwrap(); // emprunte `group`
    /// let _ = group.unregister(ring).unwrap();  // déplace `group`…
    /// let _ = buf.id();                          // …emprunt encore vivant ⇒ ERREUR
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Erreurs de `io_uring_register` ([`Errno::EINVAL`] si le groupe n'est pas
    /// enregistré). En cas d'erreur, la mémoire est tout de même libérée.
    pub fn unregister(self, ring: &mut IoUring) -> Result<(), Errno> {
        let reg = raw::IoUringBufReg {
            ring_addr: 0,
            ring_entries: 0,
            bgid: self.bgid,
            flags: 0,
            resv: [0; 3],
        };
        // SAFETY: UNREGISTER_PBUF_RING attend une `io_uring_buf_reg` (bgid,
        // nr_args == 1) ; `reg` vit pendant l'appel.
        let ret = unsafe {
            syscall::register(
                ring.fd_raw(),
                raw::IORING_UNREGISTER_PBUF_RING,
                core::ptr::from_ref(&reg) as u64,
                1,
            )
        };
        if ret < 0 {
            return Err(raw::errno_from_negative_syscall_ret(ret));
        }
        Ok(())
    }

    /// `IORING_REGISTER_PBUF_STATUS` (26) : tête courante du groupe (diagnostic /
    /// régulation).
    ///
    /// # Errors
    ///
    /// Erreurs de `io_uring_register`.
    pub fn status(&self, ring: &IoUring) -> Result<ProvidedBufferRingStatus, Errno> {
        let mut st = raw::IoUringBufStatus {
            buf_group: u32::from(self.bgid),
            head: 0,
            resv: [0; 8],
        };
        // SAFETY: PBUF_STATUS attend une `io_uring_buf_status` (nr_args == 1) ; le
        // kernel y réécrit `head`. `st` vit pendant l'appel.
        let ret = unsafe {
            syscall::register(
                ring.fd_raw(),
                raw::IORING_REGISTER_PBUF_STATUS,
                core::ptr::from_mut(&mut st) as u64,
                1,
            )
        };
        if ret < 0 {
            return Err(raw::errno_from_negative_syscall_ret(ret));
        }
        Ok(ProvidedBufferRingStatus { head: st.head })
    }

    /// Identifiant de groupe (`bgid`).
    #[must_use]
    pub fn group_id(&self) -> u16 {
        self.bgid
    }

    /// Nombre de buffers actuellement disponibles (non checkout).
    #[must_use]
    pub fn available(&self) -> u16 {
        self.available
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Soumission avec sélection automatique
// ───────────────────────────────────────────────────────────────────────────

impl IoUring {
    /// `IORING_OP_RECV` (27) avec **sélection automatique** de buffer dans
    /// `group` (`IOSQE_BUFFER_SELECT`). Aucun buffer n'est attaché au slot S1.
    /// `bundle` active `IORING_RECVSEND_BUNDLE` (plusieurs buffers en un recv).
    /// Pénurie ⇒ complétion `-ENOBUFS`.
    ///
    /// # Errors
    ///
    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins ; à la complétion,
    /// `-ENOBUFS` si le groupe est vide.
    pub fn submit_receive_provided(
        &mut self,
        sock: BorrowedFd<'_>,
        group: &ProvidedBufferRing,
        flags: MessageFlags,
        bundle: bool,
    ) -> Result<SubmissionToken, Errno> {
        let fd = sock.as_raw_fd();
        let bgid = group.group_id();
        let msg_flags = flags.bits().cast_unsigned();
        self.pending_options = self.pending_options.buffer_select();
        self.submit_op(None, |sqe| {
            sqe.opcode = raw::IORING_OP_RECV;
            sqe.fd = fd;
            sqe.len = 0; // le kernel borne par la taille du buffer choisi
            sqe.op_flags = msg_flags;
            sqe.buf_index_or_group = bgid;
            if bundle {
                sqe.ioprio |= raw::IORING_RECVSEND_BUNDLE;
            }
        })
    }

    /// `IORING_OP_READ` (22) avec sélection automatique de buffer. `length`
    /// borne le nombre d'octets lus (≤ taille du buffer). `bundle` active
    /// `IORING_RECVSEND_BUNDLE`.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_receive_provided`].
    pub fn submit_read_provided(
        &mut self,
        fd: BorrowedFd<'_>,
        group: &ProvidedBufferRing,
        length: u32,
        offset: Option<u64>,
        bundle: bool,
    ) -> Result<SubmissionToken, Errno> {
        let fd = fd.as_raw_fd();
        let bgid = group.group_id();
        let off = offset.unwrap_or(u64::MAX);
        self.pending_options = self.pending_options.buffer_select();
        self.submit_op(None, |sqe| {
            sqe.opcode = raw::IORING_OP_READ;
            sqe.fd = fd;
            sqe.len = length;
            sqe.off_or_addr2 = off;
            sqe.buf_index_or_group = bgid;
            if bundle {
                sqe.ioprio |= raw::IORING_RECVSEND_BUNDLE;
            }
        })
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Consommation : le guard ProvidedBuffer
// ───────────────────────────────────────────────────────────────────────────

impl Completion {
    /// Récupère le buffer **choisi par le kernel** pour cette complétion
    /// (`CQE_F_BUFFER`). `None` si la complétion n'a pas consommé de buffer fourni
    /// (op sans sélection, ou pénurie `-ENOBUFS`).
    ///
    /// L'emprunt `&'r mut ProvidedBufferRing` interdit le désenregistrement du
    /// groupe tant que le buffer est en cours de traitement.
    #[must_use]
    pub fn into_provided_buffer<'r>(
        self,
        group: &'r mut ProvidedBufferRing,
    ) -> Option<ProvidedBuffer<'r>> {
        let bid = self.buffer_id()?;
        let flags = self.flags();
        // Longueur utile = `cqe->res` octets reçus, bornée à la taille du buffer.
        let received = usize::try_from(self.raw_result()).unwrap_or(0);
        let len = received.min(usize::try_from(group.buf_size).expect("u32 ⊆ usize"));
        // Incrémental : `CQE_F_BUF_MORE` ⇒ le kernel garde le buffer (pas de
        // réapprovisionnement à la libération du guard).
        let replenish = !flags.contains(CompletionFlags::BUF_MORE);
        group.checkout(bid);
        Some(ProvidedBuffer {
            group,
            bid,
            len,
            replenish,
        })
    }
}

/// Accès RAII aux données reçues dans un buffer fourni. **Rend le buffer au
/// groupe** (réapprovisionnement) à la libération — sauf en mode incrémental
/// quand `CQE_F_BUF_MORE` est positionné (le kernel garde le buffer).
///
/// L'emprunt `&'r mut ProvidedBufferRing` empêche le désenregistrement du groupe
/// tant que le guard vit (sûreté par construction).
#[derive(Debug)]
pub struct ProvidedBuffer<'r> {
    /// Groupe propriétaire (emprunté mut → désenregistrement interdit).
    group: &'r mut ProvidedBufferRing,
    /// Id du buffer choisi par le kernel.
    bid: u16,
    /// Octets utiles reçus (`cqe->res`, borné à `buf_size`).
    len: usize,
    /// Réapprovisionner à la libération (faux si `CQE_F_BUF_MORE`).
    replenish: bool,
}

impl ProvidedBuffer<'_> {
    /// Id du buffer (16 bits hauts de `cqe->flags`).
    #[must_use]
    pub fn id(&self) -> u16 {
        self.bid
    }

    /// Octets reçus (longueur = `cqe->res`).
    #[must_use]
    pub fn data(&self) -> &[u8] {
        self.group.buffer_bytes(self.bid, self.len)
    }

    /// Octets reçus, en accès mutable.
    #[must_use]
    pub fn data_mut(&mut self) -> &mut [u8] {
        self.group.buffer_bytes_mut(self.bid, self.len)
    }
}

impl Drop for ProvidedBuffer<'_> {
    fn drop(&mut self) {
        if self.replenish {
            self.group.return_buffer(self.bid);
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Tests
// ───────────────────────────────────────────────────────────────────────────
//
// Tous les tests touchent `io_uring_register`/`enter` (non modélisés par Miri) →
// `#[cfg_attr(miri, ignore)]`. La logique du guard (réapprovisionnement,
// incrémental `BUF_MORE`, branche `None`) est exercée **déterministiquement** via
// des `Completion` **synthétiques** (champs construits dans ce module enfant),
// sans dépendre du timing socket ni de la mise à `BUF_MORE` par le kernel. La
// sûreté de lifetime (désenregistrement interdit sous un buffer vivant) est
// prouvée par le doctest `compile_fail` sur `unregister`.

#[cfg(test)]
mod tests {
    use super::syscall::sim::{self, Syscall};
    use super::*;
    use crate::io_uring::CompletionFlags;
    use crate::io_uring::SubmissionToken;
    use crate::test_support::{sfd, std_file};
    use air_sys_types::fd::AsFd;
    use std::io::Write;

    fn ring8() -> IoUring {
        IoUring::new(NonZeroU32::new(8).expect("8 ≠ 0")).expect("io_uring_setup 6.12")
    }

    fn nz16(n: u16) -> NonZeroU16 {
        NonZeroU16::new(n).expect("n ≠ 0")
    }

    fn nz32(n: u32) -> NonZeroU32 {
        NonZeroU32::new(n).expect("n ≠ 0")
    }

    fn tcp_pair() -> (std::net::TcpStream, std::net::TcpStream) {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback");
        let addr = listener.local_addr().expect("local_addr");
        let client = std::net::TcpStream::connect(addr).expect("connect loopback");
        let (server, _) = listener.accept().expect("accept loopback");
        (client, server)
    }

    fn complete_one(ring: &mut IoUring, expected: SubmissionToken) -> Completion {
        ring.submit_and_wait(1).expect("submit_and_wait");
        let c = ring.wait_completion().expect("wait_completion");
        assert_eq!(c.token(), expected, "user_data round-trip");
        c
    }

    /// Construit une `Completion` synthétique (id de buffer + flags + res).
    fn synth_completion(bid: u16, res: i32, buf_more: bool) -> Completion {
        let mut raw_flags = CompletionFlags::BUFFER.bits() | (u32::from(bid) << 16);
        if buf_more {
            raw_flags |= CompletionFlags::BUF_MORE.bits();
        }
        Completion {
            token: SubmissionToken::new(0, 0),
            res,
            raw_flags,
            payload: None,
            multishot: false,
        }
    }

    // ── Déterministe : logique du guard (réappro, incrémental, None) ──────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_register non supporté par Miri")]
    fn guard_replenishes_on_drop_and_exposes_data() {
        let mut ring = ring8();
        let mut group =
            ProvidedBufferRing::register(&mut ring, 7, nz16(4), nz32(64), Default::default())
                .expect("register");
        assert_eq!(group.group_id(), 7);
        assert_eq!(group.available(), 4);
        // Complétion synthétique : buffer 2, 5 octets reçus, pas d'incrémental.
        let completion = synth_completion(2, 5, false);
        {
            let mut buf = completion
                .into_provided_buffer(&mut group)
                .expect("buffer présent");
            assert_eq!(buf.id(), 2);
            assert_eq!(buf.data().len(), 5);
            buf.data_mut()[0] = 0xAB;
            assert_eq!(buf.data()[0], 0xAB);
        } // drop → réapprovisionne
        assert_eq!(group.available(), 4, "réappro après libération du guard");
        // checkout sans return (guard oublié via mem::forget) → available baisse.
        let completion = synth_completion(0, 3, false);
        let buf = completion
            .into_provided_buffer(&mut group)
            .expect("buffer présent");
        core::mem::forget(buf); // pas de réappro
        assert_eq!(group.available(), 3, "checkout non rendu");
        group.unregister(&mut ring).expect("unregister");
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_register non supporté par Miri")]
    fn incremental_buf_more_does_not_replenish() {
        let mut ring = ring8();
        let mut group = ProvidedBufferRing::register(
            &mut ring,
            3,
            nz16(2),
            nz32(64),
            ProvidedBufferRingOptions {
                kernel_mmap: false,
                incremental: true,
            },
        )
        .expect("register incremental");
        assert_eq!(group.available(), 2);
        // BUF_MORE positionné ⇒ le guard ne réapprovisionne pas (kernel garde le buffer).
        let completion = synth_completion(1, 8, true);
        {
            let buf = completion
                .into_provided_buffer(&mut group)
                .expect("buffer présent");
            assert_eq!(buf.id(), 1);
        } // drop → PAS de réappro (BUF_MORE)
        assert_eq!(
            group.available(),
            1,
            "incrémental : pas de réappro sur BUF_MORE"
        );
        // Complétion finale (sans BUF_MORE) ⇒ réappro.
        let completion = synth_completion(1, 8, false);
        drop(completion.into_provided_buffer(&mut group).expect("buffer"));
        assert_eq!(group.available(), 2, "réappro à la complétion finale");
        group.unregister(&mut ring).expect("unregister");
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_register non supporté par Miri")]
    fn into_provided_buffer_none_without_buffer_flag() {
        let mut ring = ring8();
        let mut group =
            ProvidedBufferRing::register(&mut ring, 0, nz16(2), nz32(16), Default::default())
                .expect("register");
        // Complétion sans CQE_F_BUFFER (ex. -ENOBUFS) ⇒ pas de guard.
        let completion = Completion {
            token: SubmissionToken::new(0, 0),
            res: -105, // -ENOBUFS
            raw_flags: 0,
            payload: None,
            multishot: false,
        };
        assert!(completion.into_provided_buffer(&mut group).is_none());
        assert_eq!(group.available(), 2, "available inchangé sans checkout");
        group.unregister(&mut ring).expect("unregister");
    }

    // ── Intégration kernel : recv loopback + réapprovisionnement ──────────────

    #[test]
    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
    fn provided_recv_round_trip_and_replenish() {
        let (mut client, server) = tcp_pair();
        client.write_all(b"hello").expect("write");
        client.flush().expect("flush");
        std::thread::sleep(std::time::Duration::from_millis(20));
        let mut ring = ring8();
        let mut group =
            ProvidedBufferRing::register(&mut ring, 1, nz16(4), nz32(64), Default::default())
                .expect("register");
        let tok = ring
            .submit_receive_provided(sfd(&server), &group, MessageFlags::empty(), false)
            .expect("submit recv_provided");
        let completion = complete_one(&mut ring, tok);
        let id;
        {
            let buf = completion
                .into_provided_buffer(&mut group)
                .expect("buffer choisi par le kernel");
            id = buf.id();
            assert!(id < 4);
            assert_eq!(buf.data(), b"hello");
            // (available() ne peut être lu tant que le guard emprunte `group`
            //  — la décrémentation au checkout est vérifiée par le test `forget`.)
        } // drop → réappro
        assert_eq!(group.available(), 4, "réappro après traitement");
        group.unregister(&mut ring).expect("unregister");
        let _ = id;
    }

    #[test]
    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
    fn provided_enobufs_then_replenish() {
        let (mut client, server) = tcp_pair();
        client.write_all(b"AAAAAAAABBBBBBBB").expect("write 16"); // 2 × 8
        client.flush().expect("flush");
        std::thread::sleep(std::time::Duration::from_millis(20));
        let mut ring = ring8();
        let mut group =
            ProvidedBufferRing::register(&mut ring, 2, nz16(1), nz32(8), Default::default())
                .expect("register count=1");
        // Deux recv soumis avant traitement : 1 buffer ⇒ l'un OK, l'autre -ENOBUFS.
        let t1 = ring
            .submit_receive_provided(sfd(&server), &group, MessageFlags::empty(), false)
            .expect("recv1");
        let t2 = ring
            .submit_receive_provided(sfd(&server), &group, MessageFlags::empty(), false)
            .expect("recv2");
        ring.submit_and_wait(2).expect("submit 2");
        let mut ok = 0;
        let mut enobufs = 0;
        for _ in 0..2 {
            let c = ring.wait_completion().expect("completion");
            assert!(c.token() == t1 || c.token() == t2);
            if c.flags().contains(CompletionFlags::BUFFER) {
                let buf = c.into_provided_buffer(&mut group).expect("buffer");
                assert_eq!(buf.data().len(), 8);
                ok += 1;
                drop(buf); // réappro
            } else {
                assert_eq!(c.into_result(), Err(Errno::ENOBUFS));
                enobufs += 1;
            }
        }
        assert_eq!((ok, enobufs), (1, 1), "1 servi, 1 en pénurie");
        // Réapprovisionné : un nouveau recv obtient les octets restants.
        let t3 = ring
            .submit_receive_provided(sfd(&server), &group, MessageFlags::empty(), false)
            .expect("recv3");
        let c = complete_one(&mut ring, t3);
        let buf = c
            .into_provided_buffer(&mut group)
            .expect("buffer après réappro");
        assert_eq!(buf.data(), b"BBBBBBBB");
        drop(buf);
        group.unregister(&mut ring).expect("unregister");
    }

    #[test]
    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
    fn provided_kernel_mmap_mode_round_trip() {
        let (mut client, server) = tcp_pair();
        client.write_all(b"viamap").expect("write");
        client.flush().expect("flush");
        std::thread::sleep(std::time::Duration::from_millis(20));
        let mut ring = ring8();
        let mut group = ProvidedBufferRing::register(
            &mut ring,
            5,
            nz16(4),
            nz32(64),
            ProvidedBufferRingOptions {
                kernel_mmap: true,
                incremental: false,
            },
        )
        .expect("register kernel_mmap");
        let tok = ring
            .submit_receive_provided(sfd(&server), &group, MessageFlags::empty(), false)
            .expect("recv");
        let c = complete_one(&mut ring, tok);
        let buf = c.into_provided_buffer(&mut group).expect("buffer");
        assert_eq!(buf.data(), b"viamap");
        drop(buf);
        group.unregister(&mut ring).expect("unregister");
    }

    #[test]
    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
    fn provided_status_and_read_and_bundle() {
        let mut ring = ring8();
        let mut group =
            ProvidedBufferRing::register(&mut ring, 9, nz16(4), nz32(64), Default::default())
                .expect("register");
        // status() : tête courante (0 au départ, aucun buffer consommé).
        let status = group.status(&ring).expect("status");
        assert_eq!(status.head, 0);
        // read_provided sur un pipe alimenté.
        let (read_end, write_end) = pipe_pair();
        {
            let mut w = std_file(write_end);
            w.write_all(b"piped!").expect("write pipe");
            w.flush().expect("flush");
        }
        let tok = ring
            .submit_read_provided(read_end.as_fd(), &group, 64, None, false)
            .expect("read_provided");
        let c = complete_one(&mut ring, tok);
        let buf = c.into_provided_buffer(&mut group).expect("buffer read");
        assert_eq!(buf.data(), b"piped!");
        drop(buf);
        // read_provided **bundlé** (couvre la branche `bundle` du read) : se soumet
        // et complète (le résultat exact dépend du kernel ; on n'asserte que le
        // câblage de la façade).
        let (read2, write2) = pipe_pair();
        {
            let mut w = std_file(write2);
            w.write_all(b"rd2").expect("write pipe2");
            w.flush().expect("flush");
        }
        let tok = ring
            .submit_read_provided(read2.as_fd(), &group, 64, Some(0), true)
            .expect("read_provided bundle");
        let c = complete_one(&mut ring, tok);
        if let Some(buf) = c.into_provided_buffer(&mut group) {
            drop(buf);
        }
        // bundle : un recv bundlé se soumet et complète (socket loopback).
        let (mut client, server) = tcp_pair();
        client.write_all(b"bundled").expect("write");
        client.flush().expect("flush");
        std::thread::sleep(std::time::Duration::from_millis(20));
        let tok = ring
            .submit_receive_provided(sfd(&server), &group, MessageFlags::empty(), true)
            .expect("recv bundle");
        let c = complete_one(&mut ring, tok);
        // Bundle : au moins un buffer consommé, octets présents.
        let buf = c.into_provided_buffer(&mut group).expect("buffer bundle");
        assert!(buf.data().starts_with(b"bundled") || !buf.data().is_empty());
        drop(buf);
        group.unregister(&mut ring).expect("unregister");
    }

    // ── Property-style : invariants checkout/return déterministes ─────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_register non supporté par Miri")]
    fn checkout_return_invariants() {
        let mut ring = ring8();
        let count: u16 = 4;
        let mut group =
            ProvidedBufferRing::register(&mut ring, 0, nz16(count), nz32(32), Default::default())
                .expect("register");
        // Le modèle `&mut group` du guard sérialise le traitement : un buffer à la
        // fois. On vérifie sur de nombreux cycles checkout→return (ids variés) que
        // `available` reste dans `[0, count]` et revient toujours à `count`.
        let mut returned: Vec<u16> = Vec::new();
        for step in 0u16..256 {
            assert!(group.available() <= count, "available ≤ count");
            let bid = step % count;
            let before = group.available();
            // checkout (synthétique) → traitement → return (drop).
            let c = synth_completion(bid, 4, false);
            {
                let buf = c.into_provided_buffer(&mut group).expect("buffer");
                assert_eq!(buf.id(), bid, "id rendu == id demandé, jamais mélangé");
            } // drop → réappro
            // Cycle complet : available revient à sa valeur d'avant (aucune perte).
            assert_eq!(group.available(), before, "aucun buffer perdu sur un cycle");
            returned.push(bid);
        }
        assert_eq!(group.available(), count, "tout rendu en fin");
        assert_eq!(returned.len(), 256);
        group.unregister(&mut ring).expect("unregister");
    }

    // ── Chemins d'erreur (registre injecté) ───────────────────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_register non supporté par Miri")]
    fn register_validates_and_errors_propagate() {
        sim::clear();
        let mut ring = ring8();
        // count non puissance de 2 ⇒ EINVAL en amont (aucun syscall).
        assert_eq!(
            ProvidedBufferRing::register(&mut ring, 0, nz16(3), nz32(64), Default::default()).err(),
            Some(Errno::EINVAL)
        );
        // register : erreur kernel injectée.
        sim::inject(Syscall::Register, -22);
        assert_eq!(
            ProvidedBufferRing::register(&mut ring, 0, nz16(4), nz32(64), Default::default()).err(),
            Some(Errno::EINVAL)
        );
        // Groupe réel, puis erreurs injectées sur status / unregister.
        let group =
            ProvidedBufferRing::register(&mut ring, 0, nz16(4), nz32(64), Default::default())
                .expect("register");
        sim::inject(Syscall::Register, -22);
        assert_eq!(group.status(&ring).err(), Some(Errno::EINVAL));
        sim::inject(Syscall::Register, -22);
        assert_eq!(group.unregister(&mut ring).err(), Some(Errno::EINVAL));
        sim::clear();
    }

    /// Crée une paire de pipe (lecture, écriture) via la famille `ipc`.
    fn pipe_pair() -> (air_sys_types::fd::OwnedFd, air_sys_types::fd::OwnedFd) {
        let (r, w) = crate::ipc::pipe2(air_sys_types::ipc::PipeFlags::empty()).expect("pipe2");
        (r, w)
    }
}