air_sys_syscall/io_uring/

linked.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/.

//! **Opérations liées** (Temps 3c) : exprimer qu'une opération ne démarre
//! qu'après l'achèvement de la précédente, **sans aller-retour userspace**.
//! Sous-module `air-sys-syscall::io_uring::linked`. Repose sur les drapeaux SQE
//! `IOSQE_IO_LINK` / `IOSQE_IO_HARDLINK` (axe D) et intègre l'op `LINK_TIMEOUT`
//! (15). **Aucun register opcode.**
//!
//! Référence normative : `docs/specs/layer-0/io-uring-3c-linked.md`.
//!
//! ## Sémantique soft vs hard (point critique)
//!
//! L'ordre est garanti **dans les deux cas** ; ils diffèrent par la propagation
//! des erreurs :
//!
//! - **soft** ([`LinkedChainBuilder::then`], `IOSQE_IO_LINK`) : si un maillon se
//!   termine **en erreur**, la chaîne est **rompue** — les maillons restants sont
//!   complétés avec **`-ECANCELED`**. **Définition large de l'« erreur »** : un
//!   **`read`/`recv` court** (short read — moins d'octets que demandé) **compte
//!   comme erreur** et rompt la chaîne. *Piège classique* : `read(N) → process`
//!   casse si le `read` rend `< N` octets.
//! - **hard** ([`LinkedChainBuilder::then_hard`], `IOSQE_IO_HARDLINK`) : une
//!   erreur de **complétion** d'un maillon **n'interrompt pas** (suivants
//!   exécutés) ; implique l'ordre du soft. Ne protège **pas** d'un **échec de
//!   soumission** du parent (qui rompt quand même).
//!
//! ## Staging + réservation atomique + publication unique
//!
//! Pendant la durée du builder, le `&mut IoUring` est en **mode staging** : un
//! `submit_*` appelé dans la closure réserve un slot S1 et **écrit** un SQE (avec
//! le drapeau de lien posé sur le **prédécesseur**) **sans publier la queue**. La
//! publication n'a lieu qu'au [`LinkedChainBuilder::submit`] final (**un seul
//! store-release**, Temps 1 §3.2 → le kernel voit la chaîne complète d'un coup).
//! On réutilise ainsi les `submit_*` existants **sans duplication** de `prepare_*`.
//! Si un maillon ne peut être réservé (`EBUSY`), le builder **rembobine** (libère
//! les slots déjà réservés, rembobine la `tail`) avant de retourner — **jamais de
//! chaîne partielle publiée**.

use super::owned::OwnedOp;
use super::{IoUring, SubmissionToken, raw};
use air_sys_types::Errno;
use air_sys_types::io_uring::{TimeoutFlags, TimeoutSpec};
use alloc::boxed::Box;
use alloc::vec::Vec;

impl IoUring {
    /// Démarre la construction d'une **chaîne liée** (mode staging).
    ///
    /// Les `submit_*` appelés via le builder sont **mis en attente** (slot S1
    /// réservé + SQE écrit, lien posé sur le prédécesseur) sans publier la queue ;
    /// [`LinkedChainBuilder::submit`] publie la chaîne entière en une fois.
    pub fn link_chain(&mut self) -> LinkedChainBuilder<'_> {
        let start_tail = self.sq.local_tail();
        LinkedChainBuilder {
            ring: self,
            start_tail,
            tokens: Vec::new(),
            prev_index: None,
        }
    }

    /// `IORING_OP_LINK_TIMEOUT` (15) **staged** : borne le maillon précédent. Le
    /// `__kernel_timespec` est gardé en vie dans le slot S1 (comme `submit_timeout`).
    /// Réservé au [`LinkedChainBuilder`].
    fn submit_link_timeout_staged(
        &mut self,
        spec: TimeoutSpec,
        flags: TimeoutFlags,
    ) -> Result<SubmissionToken, Errno> {
        let ts = Box::new(super::async_ops::timespec_of(spec));
        let ts_ptr = core::ptr::from_ref(&*ts) as u64;
        let op_flags = flags.bits();
        self.submit_op(Some(OwnedOp::Timeout(ts)), |sqe| {
            sqe.opcode = raw::IORING_OP_LINK_TIMEOUT;
            sqe.fd = -1;
            sqe.addr_or_splice_off_in = ts_ptr;
            sqe.len = 1;
            sqe.op_flags = op_flags;
        })
    }
}

/// Constructeur d'une **chaîne d'opérations liées** sur un [`IoUring`] en mode
/// staging (cf. module). Emprunte le ring `&mut` pour toute sa durée (aucune
/// soumission directe ne peut s'intercaler).
pub struct LinkedChainBuilder<'ring> {
    /// Ring emprunté exclusivement (staging).
    ring: &'ring mut IoUring,
    /// `tail` locale avant le premier maillon (borne de rollback).
    start_tail: u32,
    /// Jetons des maillons stagés, dans l'ordre.
    tokens: Vec<SubmissionToken>,
    /// Index (dans l'anneau SQ) du **dernier** maillon stagé, pour y poser le
    /// drapeau de lien au maillon suivant. `None` avant le premier.
    prev_index: Option<u32>,
}

impl LinkedChainBuilder<'_> {
    /// Cœur commun : stage un maillon (`op` = un `submit_*` normal). Si
    /// `link_flag` est `Some`, le pose sur le **prédécesseur**. En cas d'échec de
    /// réservation, **rembobine** la chaîne et propage l'erreur.
    fn stage<F>(mut self, link_flag: Option<u8>, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        let index = self.ring.sq.local_tail();
        match op(self.ring) {
            Ok(token) => {
                if let (Some(flag), Some(prev)) = (link_flag, self.prev_index) {
                    // SAFETY: `prev` désigne le SQE du maillon précédent, stagé
                    // pendant ce builder et **non publié** ; accès exclusif (`&mut`).
                    unsafe {
                        self.ring.sq.or_sqe_flags(prev, flag);
                    }
                }
                self.tokens.push(token);
                self.prev_index = Some(index);
                Ok(self)
            }
            Err(error) => {
                self.rollback();
                Err(error)
            }
        }
    }

    /// **Premier maillon** : `op` appelle un `submit_*` normal, mis en attente
    /// (staging). Ne pose aucun drapeau de lien (pas de prédécesseur).
    ///
    /// # Errors
    ///
    /// L'erreur du `submit_*` (ex. [`Errno::EBUSY`] si la SQ/le slab sont pleins) ;
    /// la chaîne est alors rembobinée.
    pub fn first<F>(self, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        self.stage(None, op)
    }

    /// Maillon **soft-lié** au précédent (`IOSQE_IO_LINK` posé sur le précédent) :
    /// une **erreur de complétion** du précédent (`-errno`, **short read inclus**)
    /// rompt la chaîne (suivants `-ECANCELED`).
    ///
    /// # Errors
    ///
    /// Voir [`LinkedChainBuilder::first`].
    pub fn then<F>(self, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        self.stage(Some(raw::IOSQE_IO_LINK), op)
    }

    /// Maillon **hard-lié** au précédent (`IOSQE_IO_HARDLINK` posé sur le
    /// précédent) : une erreur de complétion du précédent **n'interrompt pas** la
    /// chaîne (l'ordre reste garanti).
    ///
    /// # Errors
    ///
    /// Voir [`LinkedChainBuilder::first`].
    pub fn then_hard<F>(self, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        self.stage(Some(raw::IOSQE_IO_HARDLINK), op)
    }

    /// Borne dans le temps le **maillon précédent** via un `LINK_TIMEOUT` (op 15) :
    /// si le timeout expire avant la fin du maillon, le maillon est annulé
    /// (`-ECANCELED`) et la chaîne rompue (le timeout complète) ; si le maillon
    /// finit avant, le timeout est annulé. **Unique** manière sûre d'émettre un
    /// `LINK_TIMEOUT` (émission isolée refusée, Temps 2c).
    ///
    /// # Errors
    ///
    /// [`Errno::EINVAL`] s'il n'y a **aucun** maillon à borner ; sinon l'erreur du
    /// staging (chaîne rembobinée).
    pub fn with_link_timeout(
        mut self,
        spec: TimeoutSpec,
        flags: TimeoutFlags,
    ) -> Result<Self, Errno> {
        let Some(prev) = self.prev_index else {
            // Pas de maillon précédent à borner.
            self.rollback();
            return Err(Errno::EINVAL);
        };
        let index = self.ring.sq.local_tail();
        match self.ring.submit_link_timeout_staged(spec, flags) {
            Ok(token) => {
                // SAFETY: `prev` (maillon borné) est stagé non publié ; on le lie
                // (`IO_LINK`) au `LINK_TIMEOUT` qui le suit immédiatement. Accès
                // exclusif (`&mut`).
                unsafe {
                    self.ring.sq.or_sqe_flags(prev, raw::IOSQE_IO_LINK);
                }
                self.tokens.push(token);
                self.prev_index = Some(index);
                Ok(self)
            }
            Err(error) => {
                self.rollback();
                Err(error)
            }
        }
    }

    /// Publie la chaîne **entière, atomiquement** (un seul store-release + un
    /// `io_uring_enter`). Rend les jetons dans l'ordre de soumission.
    ///
    /// # Errors
    ///
    /// Les erreurs de [`IoUring::submit`] ([`Errno::EINTR`] remonté tel quel, etc.).
    /// La réservation des slots ayant déjà réussi au staging, l'atomicité de la
    /// chaîne est garantie en amont (jamais de chaîne partielle).
    pub fn submit(mut self) -> Result<ChainTokens, Errno> {
        self.ring.submit()?;
        Ok(ChainTokens {
            tokens: core::mem::take(&mut self.tokens),
        })
    }

    /// Rembobine le staging : libère les slots déjà réservés (idempotent côté
    /// slab) et rembobine la `tail` locale → aucun SQE de la chaîne ne sera publié.
    fn rollback(&mut self) {
        let tokens = core::mem::take(&mut self.tokens);
        for token in tokens {
            self.ring.slab.release(token);
        }
        self.ring.sq.rewind_to(self.start_tail);
        self.prev_index = None;
    }
}

/// Jetons des maillons d'une chaîne, **dans l'ordre de soumission** (corrèle
/// chaque complétion à son maillon).
#[derive(Debug, Clone)]
pub struct ChainTokens {
    tokens: Vec<SubmissionToken>,
}

impl ChainTokens {
    /// Jetons des maillons, dans l'ordre de soumission.
    #[must_use]
    pub fn tokens(&self) -> &[SubmissionToken] {
        &self.tokens
    }

    /// Jeton du maillon `index` (ordre de soumission), le cas échéant.
    #[must_use]
    pub fn get(&self, index: usize) -> Option<SubmissionToken> {
        self.tokens.get(index).copied()
    }

    /// Nombre de maillons de la chaîne.
    #[must_use]
    pub fn len(&self) -> usize {
        self.tokens.len()
    }

    /// `true` si la chaîne est vide.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.tokens.is_empty()
    }

    /// Consomme et rend les jetons possédés.
    #[must_use]
    pub fn into_vec(self) -> Vec<SubmissionToken> {
        self.tokens
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Tests
// ───────────────────────────────────────────────────────────────────────────
//
// Les chaînes touchent `io_uring_enter` (non modélisé par Miri) →
// `#[cfg_attr(miri, ignore)]`. Les accesseurs de `ChainTokens` (logique pure)
// tournent **sous Miri**. Les erreurs déterministes sont injectées via
// `submit_nop_with_result` (`IORING_NOP_INJECT_RESULT`) ; le piège du **short
// read** est testé sur un pipe réel.

#[cfg(test)]
mod tests {
    use super::*;
    use crate::io_uring::{Completion, IoUring, IoUringBuilder, SubmitOptions};
    use crate::test_support::std_file;
    use air_sys_types::fd::AsFd;
    use air_sys_types::io_uring::PollEvents;
    use air_sys_types::ipc::PipeFlags;
    use core::num::NonZeroU32;
    use core::time::Duration;
    use std::io::Write;

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

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

    fn pipe_pair() -> (air_sys_types::fd::OwnedFd, air_sys_types::fd::OwnedFd) {
        crate::ipc::pipe2(PipeFlags::empty()).expect("pipe2")
    }

    /// Draine `n` complétions et les rend **indexées par jeton** (corrélation à
    /// l'ordre de soumission via `ChainTokens`).
    fn drain(ring: &mut IoUring, n: u32) -> Vec<Completion> {
        ring.submit_and_wait(n).expect("submit_and_wait");
        let mut out = Vec::new();
        while out.len() < n as usize {
            out.push(ring.wait_completion().expect("wait_completion"));
        }
        out
    }

    /// Résultat brut de la complétion correspondant au jeton `tok`.
    fn res_of(completions: &[Completion], tok: super::SubmissionToken) -> i32 {
        completions
            .iter()
            .find(|c| c.token() == tok)
            .map(Completion::raw_result)
            .expect("complétion du jeton présente")
    }

    // ── Pur : ChainTokens (tourne sous Miri) ──────────────────────────────────

    #[test]
    fn chain_tokens_accessors() {
        let t0 = super::SubmissionToken::new(0, 0);
        let t1 = super::SubmissionToken::new(1, 0);
        let tokens = ChainTokens {
            tokens: vec![t0, t1],
        };
        assert_eq!(tokens.len(), 2);
        assert!(!tokens.is_empty());
        assert_eq!(tokens.get(0), Some(t0));
        assert_eq!(tokens.get(1), Some(t1));
        assert_eq!(tokens.get(2), None);
        assert_eq!(tokens.tokens(), &[t0, t1]);
        assert_eq!(tokens.clone().into_vec(), vec![t0, t1]);
        let empty = ChainTokens { tokens: Vec::new() };
        assert!(empty.is_empty());
        assert!(format!("{tokens:?}").contains("ChainTokens"));
    }

    // ── Intégration : ordre + complétions ─────────────────────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn chain_of_nops_completes_in_order() {
        let mut ring = ring8();
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_nop())
            .expect("first")
            .then(|r| r.submit_nop())
            .expect("then")
            .then(|r| r.submit_nop())
            .expect("then")
            .submit()
            .expect("submit");
        assert_eq!(tokens.len(), 3);
        let completions = drain(&mut ring, 3);
        for i in 0..3 {
            let tok = tokens.get(i).expect("jeton");
            assert_eq!(res_of(&completions, tok), 0, "maillon {i} succès");
        }
        assert_eq!(ring.in_flight(), 0, "aucun slot perdu");
    }

    // ── soft link : rupture sur erreur ────────────────────────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn soft_link_breaks_on_error_cancels_rest() {
        let mut ring = ring8();
        // [nop, nop_with_result(-EINVAL) soft, nop] : l'erreur du 2e rompt → 3e ECANCELED.
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_nop())
            .expect("first")
            .then(|r| r.submit_nop_with_result(-22)) // -EINVAL
            .expect("then err")
            .then(|r| r.submit_nop())
            .expect("then")
            .submit()
            .expect("submit");
        let completions = drain(&mut ring, 3);
        assert_eq!(res_of(&completions, tokens.get(0).unwrap()), 0, "1er OK");
        assert_eq!(
            res_of(&completions, tokens.get(1).unwrap()),
            -22,
            "2e erreur"
        );
        assert_eq!(
            res_of(&completions, tokens.get(2).unwrap()),
            -125,
            "3e annulé (-ECANCELED)"
        );
        assert_eq!(ring.in_flight(), 0);
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn soft_link_short_read_breaks_chain() {
        // PIÈGE : un read court (< octets demandés) rompt une chaîne soft.
        let (reader, writer) = pipe_pair();
        {
            let mut w = std_file(writer);
            w.write_all(b"short").expect("write 5"); // 5 octets dans le pipe
            w.flush().expect("flush");
        }
        let mut ring = ring8();
        // [read(64) soft, nop] : le read rend 5 (< 64) ⇒ short read ⇒ rompt → nop ECANCELED.
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_read(reader.as_fd(), vec![0u8; 64], None))
            .expect("first read")
            .then(|r| r.submit_nop())
            .expect("then")
            .submit()
            .expect("submit");
        let completions = drain(&mut ring, 2);
        let read_res = res_of(&completions, tokens.get(0).unwrap());
        let nop_res = res_of(&completions, tokens.get(1).unwrap());
        assert_eq!(read_res, 5, "read court : 5 octets lus");
        assert_eq!(
            nop_res, -125,
            "short read rompt la chaîne soft → nop -ECANCELED (piège confirmé)"
        );
        assert_eq!(ring.in_flight(), 0);
    }

    // ── hard link : l'erreur n'interrompt pas ─────────────────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn hard_link_survives_error() {
        let mut ring = ring8();
        // [nop_with_result(-EINVAL) HARD, nop] : l'erreur du 1er n'interrompt pas
        // (then_hard pose IO_HARDLINK sur le 1er) → nop s'exécute (res 0).
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_nop_with_result(-22))
            .expect("first err")
            .then_hard(|r| r.submit_nop())
            .expect("then_hard")
            .submit()
            .expect("submit");
        let completions = drain(&mut ring, 2);
        assert_eq!(
            res_of(&completions, tokens.get(0).unwrap()),
            -22,
            "1er erreur"
        );
        assert_eq!(
            res_of(&completions, tokens.get(1).unwrap()),
            0,
            "hard link : 2e exécuté malgré l'erreur"
        );
        assert_eq!(ring.in_flight(), 0);
    }

    // ── link_timeout : expirant / non expirant ────────────────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn link_timeout_expires_cancels_bounded() {
        use air_sys_types::io_uring::{TimeoutFlags, TimeoutSpec};
        // poll sur un pipe vide (jamais prêt) borné à 50 ms ⇒ poll annulé.
        let (reader, _writer) = pipe_pair();
        let mut ring = ring8();
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_poll_add(reader.as_fd(), PollEvents::IN))
            .expect("first poll")
            .with_link_timeout(
                TimeoutSpec::after(Duration::from_millis(50)),
                TimeoutFlags::empty(),
            )
            .expect("link_timeout")
            .submit()
            .expect("submit");
        let completions = drain(&mut ring, 2);
        let poll_res = res_of(&completions, tokens.get(0).unwrap());
        let timeout_res = res_of(&completions, tokens.get(1).unwrap());
        assert_eq!(poll_res, -125, "poll annulé (-ECANCELED) à l'expiration");
        // Le timeout, lui, complète (a expiré : -ETIME en 6.12).
        assert!(
            timeout_res == -62 || timeout_res >= 0,
            "timeout complète (ETIME ou succès), res = {timeout_res}"
        );
        assert_eq!(ring.in_flight(), 0);
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn link_timeout_not_expiring_is_cancelled() {
        use air_sys_types::io_uring::{TimeoutFlags, TimeoutSpec};
        let mut ring = ring8();
        // nop (instantané) borné à 5 s ⇒ le timeout est annulé.
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_nop())
            .expect("first nop")
            .with_link_timeout(
                TimeoutSpec::after(Duration::from_secs(5)),
                TimeoutFlags::empty(),
            )
            .expect("link_timeout")
            .submit()
            .expect("submit");
        let completions = drain(&mut ring, 2);
        assert_eq!(res_of(&completions, tokens.get(0).unwrap()), 0, "nop OK");
        assert_eq!(
            res_of(&completions, tokens.get(1).unwrap()),
            -125,
            "timeout annulé (maillon fini avant)"
        );
        assert_eq!(ring.in_flight(), 0);
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn with_link_timeout_without_op_is_einval() {
        let mut ring = ring8();
        use air_sys_types::io_uring::{TimeoutFlags, TimeoutSpec};
        let result = ring.link_chain().with_link_timeout(
            TimeoutSpec::after(Duration::from_secs(1)),
            TimeoutFlags::empty(),
        );
        assert!(matches!(result, Err(Errno::EINVAL)));
        // Rien n'a été stagé : un nop direct fonctionne ensuite.
        let tok = ring.submit_nop().expect("nop");
        let _ = drain(&mut ring, 1);
        let _ = tok;
        assert_eq!(ring.in_flight(), 0);
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn with_link_timeout_staging_error_rolls_back() {
        use air_sys_types::io_uring::{TimeoutFlags, TimeoutSpec};
        // slab capacité 1 : le 1er maillon consomme le slot ⇒ le LINK_TIMEOUT ne
        // peut être réservé → branche d'erreur de `with_link_timeout` + rollback.
        let mut ring = IoUringBuilder::new(nz(8))
            .max_inflight(nz(1))
            .build()
            .expect("build max_inflight 1");
        let result = ring.link_chain().first(|r| r.submit_nop()).and_then(|b| {
            b.with_link_timeout(
                TimeoutSpec::after(Duration::from_secs(1)),
                TimeoutFlags::empty(),
            )
        });
        assert!(
            matches!(result, Err(Errno::EBUSY)),
            "LINK_TIMEOUT slab plein"
        );
        assert_eq!(ring.in_flight(), 0, "rollback : slot du 1er maillon rendu");
        // Le ring reste utilisable.
        let tok = ring.submit_nop().expect("nop après rollback");
        ring.submit_and_wait(1).expect("submit_and_wait");
        assert_eq!(ring.wait_completion().expect("completion").token(), tok);
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn then_without_first_is_a_valid_unlinked_start() {
        // `then` comme tout premier maillon : pas de prédécesseur ⇒ aucun drapeau
        // de lien posé (couvre la branche `(Some(flag), None)`). Chaîne valide.
        let mut ring = ring8();
        let tokens = ring
            .link_chain()
            .then(|r| r.submit_nop())
            .expect("then sans first")
            .then(|r| r.submit_nop())
            .expect("then")
            .submit()
            .expect("submit");
        assert_eq!(tokens.len(), 2);
        let completions = drain(&mut ring, 2);
        assert_eq!(res_of(&completions, tokens.get(0).unwrap()), 0);
        assert_eq!(res_of(&completions, tokens.get(1).unwrap()), 0);
        assert_eq!(ring.in_flight(), 0);
    }

    // ── skip_cqe_on_success sur intermédiaires ────────────────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn skip_cqe_intermediate_only_final_completes() {
        let mut ring = ring8();
        let skip = || SubmitOptions::default().skip_cqe_on_success();
        let tokens = ring
            .link_chain()
            .first(|r| r.with(skip()).submit_nop())
            .expect("first skip")
            .then(|r| r.with(skip()).submit_nop())
            .expect("then skip")
            .then(|r| r.submit_nop())
            .expect("then final")
            .submit()
            .expect("submit");
        assert_eq!(tokens.len(), 3);
        // Seule la complétion finale arrive (les succès intermédiaires skippent).
        ring.submit_and_wait(1).expect("submit_and_wait");
        let final_completion = ring.wait_completion().expect("finale");
        assert_eq!(final_completion.token(), tokens.get(2).unwrap());
        assert_eq!(final_completion.raw_result(), 0);
        assert!(
            ring.try_completion().is_none(),
            "aucune complétion intermédiaire"
        );
        assert_eq!(
            ring.in_flight(),
            0,
            "slots intermédiaires libérés au staging"
        );
    }

    // ── Atomicité : slab insuffisant ⇒ EBUSY sans publication ─────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn chain_rolls_back_on_slab_exhaustion() {
        // slab capacité 2 : une chaîne de 3 ne tient pas ⇒ EBUSY au 3e, rollback.
        let mut ring = IoUringBuilder::new(nz(8))
            .max_inflight(nz(2))
            .build()
            .expect("build max_inflight 2");
        let result = ring
            .link_chain()
            .first(|r| r.submit_nop())
            .and_then(|b| b.then(|r| r.submit_nop()))
            .and_then(|b| b.then(|r| r.submit_nop())); // 3e : slab plein
        assert!(
            matches!(result, Err(Errno::EBUSY)),
            "EBUSY au maillon de trop"
        );
        // Aucun SQE publié, slab rendu : in_flight 0, un nop direct fonctionne.
        assert_eq!(ring.in_flight(), 0, "rollback : slots rendus");
        let tok = ring.submit_nop().expect("nop après rollback");
        ring.submit_and_wait(1).expect("submit_and_wait");
        let c = ring.wait_completion().expect("completion");
        assert_eq!(c.token(), tok);
        assert_eq!(ring.in_flight(), 0);
    }

    // ── Property-style : ordre = soumission, comptes cohérents ────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn chain_lengths_order_and_counts() {
        for len in 1u32..=6 {
            let mut ring = ring8();
            let mut builder = ring.link_chain();
            builder = builder.first(|r| r.submit_nop()).expect("first");
            for _ in 1..len {
                builder = builder.then(|r| r.submit_nop()).expect("then");
            }
            let tokens = builder.submit().expect("submit");
            assert_eq!(u32::try_from(tokens.len()).unwrap(), len);
            let completions = drain(&mut ring, len);
            // Chaque jeton a exactement une complétion à 0 (aucune perte/rupture).
            for i in 0..usize::try_from(len).unwrap() {
                assert_eq!(res_of(&completions, tokens.get(i).unwrap()), 0);
            }
            assert_eq!(ring.in_flight(), 0);
        }
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring_enter non supporté par Miri")]
    fn first_after_then_starts_new_unlinked_segment() {
        // `first` après un `then` : aucun drapeau de lien posé sur le prédécesseur
        // (couvre la combinaison `link_flag = None` avec `prev_index = Some`) — le
        // maillon précédent devient une queue de chaîne, un nouveau segment démarre.
        let mut ring = ring8();
        let tokens = ring
            .link_chain()
            .first(|r| r.submit_nop())
            .expect("first")
            .then(|r| r.submit_nop())
            .expect("then")
            .first(|r| r.submit_nop()) // nouveau segment, non lié au précédent
            .expect("first 2")
            .submit()
            .expect("submit");
        assert_eq!(tokens.len(), 3);
        let completions = drain(&mut ring, 3);
        for i in 0..3 {
            assert_eq!(res_of(&completions, tokens.get(i).unwrap()), 0);
        }
        assert_eq!(ring.in_flight(), 0);
    }
}