air_sys_syscall/io_uring/
net_ops.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Opérations **réseau** d'io_uring (Temps 2b), par-dessus le cœur Temps 1 et
6//! les conventions Temps 2a (ownership S1, accesseurs typés). Stratégique pour
7//! AirCom (ADR-001) : **zero-copy** du data plane et **passage de FD**
8//! (capabilities) via `sendmsg`/`recvmsg`.
9//!
10//! Référence normative : `docs/specs/layer-0/io-uring-2b-network.md`.
11//!
12//! **Invariants Air** : `MSG_NOSIGNAL` par défaut sur tout envoi (EPIPE en
13//! complétion, jamais de `SIGPIPE`), `SOCK_CLOEXEC`/`CLOEXEC` par défaut sur les
14//! FD créés (`socket`/`accept`) et les FD reçus (`MSG_CMSG_CLOEXEC`).
15//!
16//! **Hors périmètre** : variantes direct-descriptor (`accept_direct`,
17//! `socket_direct`) → Temps 3a ; multishot → Temps 3d ; buffers fournis /
18//! bundle → Temps 3b.
19//!
20//! **Cycle zero-copy à deux complétions** (§4.1) : `send_zc`/`sendmsg_zc`
21//! produisent une CQE **résultat** (`res ≥ 0` + `F_MORE`, sans restitution) puis
22//! une CQE **NOTIF** (`F_NOTIF`) qui restitue le buffer et libère le slot. C'est
23//! l'unique cas où un slot survit à sa première complétion — porté par le
24//! mécanisme `has_more` du slab (Temps 1), sans extension.
25
26use super::owned::{OwnedOp, RecvMsgState, SendMsgState};
27use super::raw::{self, Iovec, Msghdr};
28use super::{IoUring, SubmissionToken};
29use crate::net::{MAX_SOCKADDR_LEN, RawSockaddrStorage, build_scm_rights_cmsg, socket_addr_to_raw};
30use air_sys_types::Errno;
31use air_sys_types::fd::{AsFd, AsRawFd, BorrowedFd};
32use air_sys_types::net::{
33    AcceptFlags, MessageFlags, OwnedReceiveMessage, OwnedSendMessage, ShutdownMode, SocketAddr,
34    SocketDomain, SocketType, ZeroCopyFlags,
35};
36use alloc::boxed::Box;
37use alloc::vec::Vec;
38
39/// `usize` → `u64` saturant (longueurs ; jamais déclenché sur cible LP64 où
40/// `usize == u64` — `unwrap_or` n'introduit pas de branche d'erreur morte).
41fn u64_of(n: usize) -> u64 {
42    u64::try_from(n).unwrap_or(u64::MAX)
43}
44
45/// `usize` → `u32` du champ `len` du SQE, `EINVAL` si débordement (validation
46/// amont, Principe 4 — un buffer réseau > 4 GiB).
47fn len_u32(n: usize) -> Result<u32, Errno> {
48    u32::try_from(n).map_err(|_| Errno::EINVAL)
49}
50
51/// Sérialise une [`SocketAddr`] en stockage `sockaddr` **possédé** (boxé, à
52/// adresse stable) + sa longueur effective, réutilisant la sérialisation de la
53/// famille `net` (DRY).
54fn owned_sockaddr(address: &SocketAddr) -> (Box<RawSockaddrStorage>, u32) {
55    let (storage, len) = socket_addr_to_raw(address);
56    (Box::new(storage), len)
57}
58
59/// Construit l'état `sendmsg` possédé : `msghdr` + `iovec` + adresse + cmsg
60/// `SCM_RIGHTS`, chacun dans une allocation indépendante à adresse stable. Pose
61/// `msg_flags` dans le `msghdr`. Retourne l'état boxé **et** le pointeur du
62/// `msghdr` (capturé avant déplacement dans le slot).
63fn build_send_state(request: OwnedSendMessage, msg_flags: i32) -> (Box<SendMsgState>, u64) {
64    let OwnedSendMessage {
65        mut buffers,
66        address,
67        fds,
68        flags: _,
69    } = request;
70    // iovec depuis les tas (stables) des buffers possédés.
71    let mut iovec_vec: Vec<Iovec> = Vec::with_capacity(buffers.len());
72    for buf in &mut buffers {
73        iovec_vec.push(Iovec {
74            iov_base: buf.as_mut_ptr(),
75            iov_len: buf.len(),
76        });
77    }
78    let iovecs = iovec_vec.into_boxed_slice();
79    let iov_ptr = iovecs.as_ptr() as u64;
80    let iov_len = u64_of(iovecs.len());
81    // Adresse de destination possédée (optionnelle).
82    let addr = address.as_ref().map(owned_sockaddr);
83    let (name_ptr, name_len) = addr
84        .as_ref()
85        .map_or((0u64, 0u32), |(s, l)| (s.bytes.as_ptr() as u64, *l));
86    // cmsg SCM_RIGHTS depuis les FD (le buffer d'octets est possédé/boxé).
87    let fd_borrows: Vec<BorrowedFd<'_>> = fds.iter().map(AsFd::as_fd).collect();
88    let control = build_scm_rights_cmsg(&fd_borrows).into_boxed_slice();
89    let (control_ptr, control_len) = if control.is_empty() {
90        (0u64, 0u64)
91    } else {
92        (control.as_ptr() as u64, u64_of(control.len()))
93    };
94    let msghdr = Box::new(Msghdr {
95        msg_name: name_ptr,
96        msg_namelen: name_len,
97        pad1: 0,
98        msg_iov: iov_ptr,
99        msg_iovlen: iov_len,
100        msg_control: control_ptr,
101        msg_controllen: control_len,
102        msg_flags,
103        pad2: 0,
104    });
105    let hdr_ptr = core::ptr::from_ref(&*msghdr) as u64;
106    let state = Box::new(SendMsgState {
107        msghdr,
108        iovecs,
109        buffers,
110        addr: addr.map(|(s, _)| s),
111        control,
112        fds,
113    });
114    (state, hdr_ptr)
115}
116
117impl IoUring {
118    // ── Établissement de connexion ────────────────────────────────────────
119
120    /// Soumet un `IORING_OP_ACCEPT` (13) sans capturer l'adresse pair.
121    /// `CLOEXEC` est ajouté d'office. Complétion via
122    /// [`Completion::accepted_fd`](super::Completion::accepted_fd).
123    ///
124    /// # Errors
125    ///
126    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
127    pub fn submit_accept(
128        &mut self,
129        listener: BorrowedFd<'_>,
130        flags: AcceptFlags,
131    ) -> Result<SubmissionToken, Errno> {
132        let fd = listener.as_raw_fd();
133        let accept_flags = (flags | AcceptFlags::CLOEXEC).bits().cast_unsigned();
134        self.submit_op(None, |sqe| {
135            sqe.opcode = raw::IORING_OP_ACCEPT;
136            sqe.fd = fd;
137            sqe.op_flags = accept_flags;
138        })
139    }
140
141    /// Soumet un `IORING_OP_ACCEPT` (13) en capturant l'adresse pair dans un
142    /// stockage possédé du slot. Complétion via
143    /// [`Completion::into_accept_result`](super::Completion::into_accept_result).
144    ///
145    /// # Errors
146    ///
147    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
148    pub fn submit_accept_with_peer(
149        &mut self,
150        listener: BorrowedFd<'_>,
151        flags: AcceptFlags,
152    ) -> Result<SubmissionToken, Errno> {
153        let fd = listener.as_raw_fd();
154        let accept_flags = (flags | AcceptFlags::CLOEXEC).bits().cast_unsigned();
155        let addr = Box::new(RawSockaddrStorage {
156            bytes: [0u8; MAX_SOCKADDR_LEN],
157        });
158        let addrlen = Box::new(u32::try_from(MAX_SOCKADDR_LEN).unwrap_or(0));
159        let addr_ptr = addr.bytes.as_ptr() as u64;
160        let addrlen_ptr = core::ptr::from_ref(&*addrlen) as u64;
161        self.submit_op(Some(OwnedOp::Accept { addr, addrlen }), |sqe| {
162            sqe.opcode = raw::IORING_OP_ACCEPT;
163            sqe.fd = fd;
164            sqe.addr_or_splice_off_in = addr_ptr;
165            sqe.off_or_addr2 = addrlen_ptr;
166            sqe.op_flags = accept_flags;
167        })
168    }
169
170    /// Soumet un `IORING_OP_CONNECT` (16). L'adresse est sérialisée et déplacée
171    /// dans le slot. Complétion via
172    /// [`Completion::completed`](super::Completion::completed).
173    ///
174    /// # Errors
175    ///
176    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
177    pub fn submit_connect(
178        &mut self,
179        sock: BorrowedFd<'_>,
180        address: SocketAddr,
181    ) -> Result<SubmissionToken, Errno> {
182        let fd = sock.as_raw_fd();
183        let (storage, len) = owned_sockaddr(&address);
184        let addr_ptr = storage.bytes.as_ptr() as u64;
185        self.submit_op(Some(OwnedOp::SockAddr(storage)), |sqe| {
186            sqe.opcode = raw::IORING_OP_CONNECT;
187            sqe.fd = fd;
188            sqe.addr_or_splice_off_in = addr_ptr;
189            sqe.off_or_addr2 = u64::from(len);
190        })
191    }
192
193    /// Soumet un `IORING_OP_SOCKET` (45). `SOCK_CLOEXEC` est ajouté d'office.
194    /// Complétion via
195    /// [`Completion::into_socket_fd`](super::Completion::into_socket_fd).
196    ///
197    /// # Errors
198    ///
199    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
200    pub fn submit_socket(
201        &mut self,
202        domain: SocketDomain,
203        ty: SocketType,
204        protocol: i32,
205    ) -> Result<SubmissionToken, Errno> {
206        let domain = domain as i32;
207        let ty = (ty as i32) | raw::SOCK_CLOEXEC;
208        let protocol = protocol.cast_unsigned();
209        self.submit_op(None, |sqe| {
210            sqe.opcode = raw::IORING_OP_SOCKET;
211            sqe.fd = domain;
212            sqe.off_or_addr2 = u64::from(ty.cast_unsigned());
213            sqe.len = protocol;
214        })
215    }
216
217    /// Soumet un `IORING_OP_BIND` (56, kernel ≥ 6.11). Complétion via `completed`.
218    ///
219    /// # Errors
220    ///
221    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
222    pub fn submit_bind(
223        &mut self,
224        sock: BorrowedFd<'_>,
225        address: SocketAddr,
226    ) -> Result<SubmissionToken, Errno> {
227        let fd = sock.as_raw_fd();
228        let (storage, len) = owned_sockaddr(&address);
229        let addr_ptr = storage.bytes.as_ptr() as u64;
230        self.submit_op(Some(OwnedOp::SockAddr(storage)), |sqe| {
231            sqe.opcode = raw::IORING_OP_BIND;
232            sqe.fd = fd;
233            sqe.addr_or_splice_off_in = addr_ptr;
234            sqe.off_or_addr2 = u64::from(len);
235        })
236    }
237
238    /// Soumet un `IORING_OP_LISTEN` (57, kernel ≥ 6.11). Complétion via
239    /// `completed`.
240    ///
241    /// # Errors
242    ///
243    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
244    pub fn submit_listen(
245        &mut self,
246        sock: BorrowedFd<'_>,
247        backlog: u32,
248    ) -> Result<SubmissionToken, Errno> {
249        let fd = sock.as_raw_fd();
250        self.submit_op(None, |sqe| {
251            sqe.opcode = raw::IORING_OP_LISTEN;
252            sqe.fd = fd;
253            sqe.len = backlog;
254        })
255    }
256
257    // ── Transfert one-shot ────────────────────────────────────────────────
258
259    /// Soumet un `IORING_OP_SEND` (26). `MSG_NOSIGNAL` ajouté par défaut. Buffer
260    /// déplacé dans le slot ; complétion via
261    /// [`Completion::into_buffer_result`](super::Completion::into_buffer_result).
262    ///
263    /// # Errors
264    ///
265    /// [`Errno::EBUSY`] (SQ/slab plein) ; [`Errno::EINVAL`] si `buffer.len()`
266    /// déborde un `u32`.
267    pub fn submit_send(
268        &mut self,
269        sock: BorrowedFd<'_>,
270        mut buffer: Vec<u8>,
271        flags: MessageFlags,
272    ) -> Result<SubmissionToken, Errno> {
273        let fd = sock.as_raw_fd();
274        let len = len_u32(buffer.len())?;
275        let addr = buffer.as_mut_ptr() as u64;
276        let msg_flags = (flags | MessageFlags::NOSIGNAL).bits().cast_unsigned();
277        self.submit_op(Some(OwnedOp::Bytes(buffer)), |sqe| {
278            sqe.opcode = raw::IORING_OP_SEND;
279            sqe.fd = fd;
280            sqe.addr_or_splice_off_in = addr;
281            sqe.len = len;
282            sqe.op_flags = msg_flags;
283        })
284    }
285
286    /// Soumet un `IORING_OP_RECV` (27). Buffer déplacé dans le slot ; complétion
287    /// via `into_buffer_result`. `socket_has_pending_data()` (CQE_F_SOCK_NONEMPTY)
288    /// indique s'il reste des données à lire.
289    ///
290    /// # Errors
291    ///
292    /// [`Errno::EBUSY`] (SQ/slab plein) ; [`Errno::EINVAL`] si `buffer.len()`
293    /// déborde un `u32`.
294    pub fn submit_receive(
295        &mut self,
296        sock: BorrowedFd<'_>,
297        mut buffer: Vec<u8>,
298        flags: MessageFlags,
299    ) -> Result<SubmissionToken, Errno> {
300        let fd = sock.as_raw_fd();
301        let len = len_u32(buffer.len())?;
302        let addr = buffer.as_mut_ptr() as u64;
303        let msg_flags = flags.bits().cast_unsigned();
304        self.submit_op(Some(OwnedOp::Bytes(buffer)), |sqe| {
305            sqe.opcode = raw::IORING_OP_RECV;
306            sqe.fd = fd;
307            sqe.addr_or_splice_off_in = addr;
308            sqe.len = len;
309            sqe.op_flags = msg_flags;
310        })
311    }
312
313    /// Soumet un `IORING_OP_SENDMSG` (9) : envoi vectorisé + **passage de FD**
314    /// (`SCM_RIGHTS`). `MSG_NOSIGNAL` par défaut. Tout l'état est déplacé dans le
315    /// slot. Complétion via
316    /// [`Completion::into_result`](super::Completion::into_result) (octets).
317    ///
318    /// # Errors
319    ///
320    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
321    pub fn submit_send_message(
322        &mut self,
323        sock: BorrowedFd<'_>,
324        request: OwnedSendMessage,
325    ) -> Result<SubmissionToken, Errno> {
326        let fd = sock.as_raw_fd();
327        let msg_flags = (request.flags | MessageFlags::NOSIGNAL).bits();
328        let (state, hdr_ptr) = build_send_state(request, msg_flags);
329        self.submit_op(Some(OwnedOp::SendMsg(state)), |sqe| {
330            sqe.opcode = raw::IORING_OP_SENDMSG;
331            sqe.fd = fd;
332            sqe.addr_or_splice_off_in = hdr_ptr;
333            sqe.len = 1;
334        })
335    }
336
337    /// Soumet un `IORING_OP_RECVMSG` (10) : réception vectorisée + **réception de
338    /// FD** (`SCM_RIGHTS`). `MSG_CMSG_CLOEXEC` par défaut (FD reçus CLOEXEC).
339    /// Complétion via [`Completion::into_receive_message_result`](super::Completion::into_receive_message_result).
340    ///
341    /// # Errors
342    ///
343    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
344    pub fn submit_receive_message(
345        &mut self,
346        sock: BorrowedFd<'_>,
347        request: OwnedReceiveMessage,
348    ) -> Result<SubmissionToken, Errno> {
349        let fd = sock.as_raw_fd();
350        let msg_flags = (request.flags | MessageFlags::CMSG_CLOEXEC).bits();
351        let OwnedReceiveMessage {
352            mut buffers,
353            control_capacity,
354            flags: _,
355        } = request;
356        let mut iovec_vec: Vec<Iovec> = Vec::with_capacity(buffers.len());
357        for buf in &mut buffers {
358            iovec_vec.push(Iovec {
359                iov_base: buf.as_mut_ptr(),
360                iov_len: buf.len(),
361            });
362        }
363        let iovecs = iovec_vec.into_boxed_slice();
364        let iov_ptr = iovecs.as_ptr() as u64;
365        let iov_len = u64_of(iovecs.len());
366        let name = Box::new(RawSockaddrStorage {
367            bytes: [0u8; MAX_SOCKADDR_LEN],
368        });
369        let name_ptr = name.bytes.as_ptr() as u64;
370        let control = vec![0u8; control_capacity].into_boxed_slice();
371        let (control_ptr, control_len) = if control.is_empty() {
372            (0u64, 0u64)
373        } else {
374            (control.as_ptr() as u64, u64_of(control.len()))
375        };
376        let name_capacity = u32::try_from(MAX_SOCKADDR_LEN).unwrap_or(0);
377        let msghdr = Box::new(Msghdr {
378            msg_name: name_ptr,
379            msg_namelen: name_capacity,
380            pad1: 0,
381            msg_iov: iov_ptr,
382            msg_iovlen: iov_len,
383            msg_control: control_ptr,
384            msg_controllen: control_len,
385            msg_flags,
386            pad2: 0,
387        });
388        let hdr_ptr = core::ptr::from_ref(&*msghdr) as u64;
389        let state = Box::new(RecvMsgState {
390            msghdr,
391            iovecs,
392            buffers,
393            name,
394            control,
395        });
396        self.submit_op(Some(OwnedOp::RecvMsg(state)), |sqe| {
397            sqe.opcode = raw::IORING_OP_RECVMSG;
398            sqe.fd = fd;
399            sqe.addr_or_splice_off_in = hdr_ptr;
400            sqe.len = 1;
401        })
402    }
403
404    // ── Zero-copy (data plane AirCom) ─────────────────────────────────────
405
406    /// Soumet un `IORING_OP_SEND_ZC` (47). **Deux complétions** (§4.1) : résultat
407    /// (`F_MORE`, `into_result`) puis NOTIF (`into_zero_copy_buffer`). Le buffer
408    /// reste vivant **jusqu'au NOTIF**. `MSG_NOSIGNAL` par défaut.
409    ///
410    /// # Errors
411    ///
412    /// [`Errno::EBUSY`] (SQ/slab plein) ; [`Errno::EINVAL`] si `buffer.len()`
413    /// déborde un `u32`.
414    pub fn submit_send_zero_copy(
415        &mut self,
416        sock: BorrowedFd<'_>,
417        mut buffer: Vec<u8>,
418        flags: MessageFlags,
419        zero_copy: ZeroCopyFlags,
420    ) -> Result<SubmissionToken, Errno> {
421        let fd = sock.as_raw_fd();
422        let len = len_u32(buffer.len())?;
423        let addr = buffer.as_mut_ptr() as u64;
424        let msg_flags = (flags | MessageFlags::NOSIGNAL).bits().cast_unsigned();
425        let zc = u16::try_from(zero_copy.bits()).unwrap_or(raw::IORING_SEND_ZC_REPORT_USAGE);
426        self.submit_op(Some(OwnedOp::Bytes(buffer)), |sqe| {
427            sqe.opcode = raw::IORING_OP_SEND_ZC;
428            sqe.fd = fd;
429            sqe.addr_or_splice_off_in = addr;
430            sqe.len = len;
431            sqe.op_flags = msg_flags;
432            sqe.ioprio = zc;
433        })
434    }
435
436    /// Soumet un `IORING_OP_SENDMSG_ZC` (48) : `sendmsg` zero-copy (vectorisé +
437    /// FD passing). Deux complétions (§4.1). `MSG_NOSIGNAL` par défaut.
438    ///
439    /// # Errors
440    ///
441    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
442    pub fn submit_send_message_zero_copy(
443        &mut self,
444        sock: BorrowedFd<'_>,
445        request: OwnedSendMessage,
446        zero_copy: ZeroCopyFlags,
447    ) -> Result<SubmissionToken, Errno> {
448        let fd = sock.as_raw_fd();
449        let msg_flags = (request.flags | MessageFlags::NOSIGNAL).bits();
450        let (state, hdr_ptr) = build_send_state(request, msg_flags);
451        let zc = u16::try_from(zero_copy.bits()).unwrap_or(raw::IORING_SEND_ZC_REPORT_USAGE);
452        self.submit_op(Some(OwnedOp::SendMsg(state)), |sqe| {
453            sqe.opcode = raw::IORING_OP_SENDMSG_ZC;
454            sqe.fd = fd;
455            sqe.addr_or_splice_off_in = hdr_ptr;
456            sqe.len = 1;
457            sqe.ioprio = zc;
458        })
459    }
460
461    // ── Fermeture ─────────────────────────────────────────────────────────
462
463    /// Soumet un `IORING_OP_SHUTDOWN` (34). Complétion via `completed`.
464    ///
465    /// # Errors
466    ///
467    /// [`Errno::EBUSY`] si la SQ ou le slab sont pleins.
468    pub fn submit_shutdown(
469        &mut self,
470        sock: BorrowedFd<'_>,
471        how: ShutdownMode,
472    ) -> Result<SubmissionToken, Errno> {
473        let fd = sock.as_raw_fd();
474        let how = (how as i32).cast_unsigned();
475        self.submit_op(None, |sqe| {
476            sqe.opcode = raw::IORING_OP_SHUTDOWN;
477            sqe.fd = fd;
478            sqe.len = how;
479        })
480    }
481}
482
483// ───────────────────────────────────────────────────────────────────────────
484// Tests
485// ───────────────────────────────────────────────────────────────────────────
486//
487// Intégration (kernel 6.12 réel : AF_UNIX socketpair + AF_UNIX path) :
488// `#[cfg_attr(miri, ignore)]` (Miri ne modélise pas `io_uring_enter`). Les tests
489// purs (accesseurs, ownership zero-copy/recvmsg) tournent SOUS Miri.
490
491#[cfg(test)]
492mod tests {
493    use super::*;
494    use crate::io_uring::{Completion, IoUring, SubmissionToken};
495    use crate::test_support::{sfd, sown, std_file};
496    use air_sys_types::net::UnixSocketAddr;
497    use core::num::NonZeroU32;
498    use std::ffi::CString;
499    use std::io::{Read, Write};
500    use std::os::unix::net::UnixStream;
501    use std::sync::atomic::{AtomicU32, Ordering};
502
503    static COUNTER: AtomicU32 = AtomicU32::new(0);
504
505    fn ring8() -> IoUring {
506        IoUring::new(NonZeroU32::new(16).expect("16 ≠ 0")).expect("io_uring_setup 6.12")
507    }
508
509    fn pair() -> (UnixStream, UnixStream) {
510        UnixStream::pair().expect("socketpair AF_UNIX")
511    }
512
513    /// Paire TCP connectée sur loopback (sockets capables de zero-copy).
514    fn tcp_pair() -> (std::net::TcpStream, std::net::TcpStream) {
515        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback");
516        let addr = listener.local_addr().expect("local_addr");
517        let client = std::net::TcpStream::connect(addr).expect("connect loopback");
518        let (server, _) = listener.accept().expect("accept loopback");
519        (client, server)
520    }
521
522    fn complete_one(ring: &mut IoUring, expected: SubmissionToken) -> Completion {
523        ring.submit_and_wait(1).expect("submit_and_wait");
524        let c = ring.wait_completion().expect("wait_completion");
525        assert_eq!(c.token(), expected, "user_data round-trip");
526        c
527    }
528
529    fn unix_path(tag: &str) -> (CString, std::path::PathBuf) {
530        let n = COUNTER.fetch_add(1, Ordering::Relaxed);
531        let p = std::env::temp_dir().join(format!("air-iou2b-{}-{n}-{tag}", std::process::id()));
532        let _ = std::fs::remove_file(&p);
533        let c = CString::new(p.as_os_str().as_encoded_bytes()).expect("chemin sans NUL");
534        (c, p)
535    }
536
537    // ── Intégration : send / recv one-shot ────────────────────────────────
538
539    #[test]
540    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
541    fn send_recv_round_trip() {
542        let (a, b) = pair();
543        let mut ring = ring8();
544        let tok = ring
545            .submit_send(sfd(&a), b"bonjour".to_vec(), MessageFlags::empty())
546            .expect("submit_send");
547        let (_buf, n) = complete_one(&mut ring, tok)
548            .into_buffer_result()
549            .expect("send ok");
550        assert_eq!(n, 7);
551
552        let tok = ring
553            .submit_receive(sfd(&b), vec![0u8; 16], MessageFlags::empty())
554            .expect("submit_receive");
555        let cmp = complete_one(&mut ring, tok);
556        let _pending = cmp.socket_has_pending_data();
557        let (buf, n) = cmp.into_buffer_result().expect("recv ok");
558        assert_eq!(&buf[..n], b"bonjour");
559    }
560
561    // ── Intégration : passage de FD (SCM_RIGHTS) — cœur AirCom ─────────────
562
563    #[test]
564    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
565    fn fd_passing_round_trip_same_object() {
566        let (a, b) = pair();
567        let mut ring = ring8();
568        // FD à passer : l'extrémité écriture d'un pipe ; on garde la lecture.
569        let (mut reader, writer) = std::io::pipe().expect("pipe");
570        let send = OwnedSendMessage {
571            buffers: vec![b"cap".to_vec()],
572            address: None,
573            fds: vec![sown(writer)],
574            flags: MessageFlags::empty(),
575        };
576        let tok = ring
577            .submit_send_message(sfd(&a), send)
578            .expect("submit_send_message");
579        let bytes = complete_one(&mut ring, tok)
580            .into_result()
581            .expect("sendmsg ok");
582        assert_eq!(bytes, 3);
583
584        let recv = OwnedReceiveMessage {
585            buffers: vec![vec![0u8; 16]],
586            control_capacity: 64,
587            flags: MessageFlags::empty(),
588        };
589        let tok = ring
590            .submit_receive_message(sfd(&b), recv)
591            .expect("submit_receive_message");
592        let (msg, meta) = complete_one(&mut ring, tok)
593            .into_receive_message_result()
594            .expect("recvmsg ok");
595        assert_eq!(meta.payloadlen, 3);
596        assert_eq!(&msg.buffers[0][..3], b"cap");
597        assert!(!meta.control_truncated(), "cmsg non tronqué");
598        assert_eq!(meta.fds.len(), 1, "un FD reçu via SCM_RIGHTS");
599
600        // Le FD reçu pointe le **même objet** : écrire au travers, lire côté local.
601        let mut received = std_file(meta.fds.into_iter().next().expect("1 fd"));
602        received.write_all(b"Z").expect("write via received fd");
603        drop(received);
604        let mut got = [0u8; 1];
605        reader.read_exact(&mut got).expect("lecture pipe local");
606        assert_eq!(&got, b"Z", "FD reçu = même objet pipe");
607    }
608
609    #[test]
610    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
611    fn fd_passing_ctrunc_no_leak() {
612        let (a, b) = pair();
613        let mut ring = ring8();
614        let (_r1, w1) = std::io::pipe().expect("pipe1");
615        let (_r2, w2) = std::io::pipe().expect("pipe2");
616        let send = OwnedSendMessage {
617            buffers: vec![b"x".to_vec()],
618            address: None,
619            fds: vec![sown(w1), sown(w2)],
620            flags: MessageFlags::empty(),
621        };
622        let tok = ring
623            .submit_send_message(sfd(&a), send)
624            .expect("submit_send_message");
625        let _ = complete_one(&mut ring, tok)
626            .into_result()
627            .expect("sendmsg ok");
628
629        // Buffer de contrôle trop petit pour 2 FD ⇒ MSG_CTRUNC, FD partiels
630        // **fermés par le kernel** (jamais matérialisés ⇒ aucune fuite).
631        let recv = OwnedReceiveMessage {
632            buffers: vec![vec![0u8; 8]],
633            control_capacity: 20, // < 16 (header) + 2*4 aligné
634            flags: MessageFlags::empty(),
635        };
636        let tok = ring
637            .submit_receive_message(sfd(&b), recv)
638            .expect("submit_receive_message");
639        let (_msg, meta) = complete_one(&mut ring, tok)
640            .into_receive_message_result()
641            .expect("recvmsg ok");
642        assert!(meta.control_truncated(), "MSG_CTRUNC attendu");
643        assert!(meta.fds.len() <= 1, "au plus un FD complet matérialisé");
644        // Les FD reçus (le cas échéant) sont des OwnedFd valides — pas d'entier brut.
645        for fd in &meta.fds {
646            assert!(fd.as_fd().as_raw_fd() >= 0);
647        }
648    }
649
650    #[test]
651    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
652    fn sendmsg_recvmsg_without_control_buffer() {
653        // `control_capacity = 0` : recvmsg sans réception de FD (couvre le bras
654        // `control.is_empty()`). sendmsg sans `fds` (cmsg vide).
655        let (a, b) = pair();
656        let mut ring = ring8();
657        let send = OwnedSendMessage {
658            buffers: vec![b"data".to_vec()],
659            address: None,
660            fds: Vec::new(),
661            flags: MessageFlags::empty(),
662        };
663        let tok = ring
664            .submit_send_message(sfd(&a), send)
665            .expect("submit_send_message");
666        let _ = complete_one(&mut ring, tok)
667            .into_result()
668            .expect("sendmsg ok");
669
670        let recv = OwnedReceiveMessage {
671            buffers: vec![vec![0u8; 16]],
672            control_capacity: 0,
673            flags: MessageFlags::empty(),
674        };
675        let tok = ring
676            .submit_receive_message(sfd(&b), recv)
677            .expect("submit_receive_message");
678        let (msg, meta) = complete_one(&mut ring, tok)
679            .into_receive_message_result()
680            .expect("recvmsg ok");
681        assert_eq!(meta.payloadlen, 4);
682        assert_eq!(&msg.buffers[0][..4], b"data");
683        assert!(meta.fds.is_empty());
684    }
685
686    // ── Intégration : socket / bind / listen / connect / accept / shutdown ─
687
688    #[test]
689    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
690    fn socket_bind_listen_connect_accept_shutdown_chain() {
691        let (addr_c, path) = unix_path("srv");
692        let addr = SocketAddr::Unix(UnixSocketAddr::Path(addr_c));
693        let mut ring = ring8();
694
695        // Listener : socket + bind + listen.
696        let tok = ring
697            .submit_socket(SocketDomain::Unix, SocketType::Stream, 0)
698            .expect("submit_socket");
699        let listener = complete_one(&mut ring, tok)
700            .into_socket_fd()
701            .expect("socket ok");
702        let tok = ring
703            .submit_bind(listener.as_fd(), addr.clone())
704            .expect("submit_bind");
705        complete_one(&mut ring, tok).completed().expect("bind ok");
706        let tok = ring
707            .submit_listen(listener.as_fd(), 8)
708            .expect("submit_listen");
709        complete_one(&mut ring, tok).completed().expect("listen ok");
710
711        // Client : socket + connect (mis en file dans le backlog).
712        let tok = ring
713            .submit_socket(SocketDomain::Unix, SocketType::Stream, 0)
714            .expect("submit_socket client");
715        let client = complete_one(&mut ring, tok)
716            .into_socket_fd()
717            .expect("socket client ok");
718        let tok = ring
719            .submit_connect(client.as_fd(), addr)
720            .expect("submit_connect");
721        complete_one(&mut ring, tok)
722            .completed()
723            .expect("connect ok");
724
725        // accept_with_peer.
726        let tok = ring
727            .submit_accept_with_peer(listener.as_fd(), AcceptFlags::empty())
728            .expect("submit_accept_with_peer");
729        let (accepted, _peer) = complete_one(&mut ring, tok)
730            .into_accept_result()
731            .expect("accept ok");
732
733        // Round-trip sur la connexion acceptée.
734        let tok = ring
735            .submit_send(client.as_fd(), b"hi".to_vec(), MessageFlags::empty())
736            .expect("send");
737        let _ = complete_one(&mut ring, tok)
738            .into_buffer_result()
739            .expect("ok");
740        let tok = ring
741            .submit_receive(accepted.as_fd(), vec![0u8; 8], MessageFlags::empty())
742            .expect("recv");
743        let (buf, n) = complete_one(&mut ring, tok)
744            .into_buffer_result()
745            .expect("ok");
746        assert_eq!(&buf[..n], b"hi");
747
748        // shutdown.
749        let tok = ring
750            .submit_shutdown(client.as_fd(), ShutdownMode::Both)
751            .expect("submit_shutdown");
752        complete_one(&mut ring, tok)
753            .completed()
754            .expect("shutdown ok");
755
756        let _ = std::fs::remove_file(&path);
757    }
758
759    #[test]
760    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
761    fn plain_accept_returns_fd() {
762        // accept (sans peer) via std listener + connexion synchrone.
763        let (_addr_c, path) = unix_path("acc");
764        let listener = std::os::unix::net::UnixListener::bind(&path).expect("bind std");
765        let _client = UnixStream::connect(&path).expect("connect std");
766        let mut ring = ring8();
767        let tok = ring
768            .submit_accept(sfd(&listener), AcceptFlags::empty())
769            .expect("submit_accept");
770        let accepted = complete_one(&mut ring, tok)
771            .accepted_fd()
772            .expect("accept ok");
773        assert!(accepted.as_fd().as_raw_fd() >= 0);
774        let _ = std::fs::remove_file(&path);
775    }
776
777    // ── Intégration : zero-copy à deux complétions ────────────────────────
778
779    #[test]
780    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
781    fn zero_copy_two_completion_cycle() {
782        // Zero-copy n'est supporté que sur des sockets capables (TCP loopback) ;
783        // AF_UNIX rendrait EOPNOTSUPP (cf. `send_zc_early_failure`).
784        let (client, mut server) = tcp_pair();
785        let mut ring = ring8();
786        let data = vec![42u8; 4096];
787        let tok = ring
788            .submit_send_zero_copy(
789                sfd(&client),
790                data.clone(),
791                MessageFlags::empty(),
792                ZeroCopyFlags::REPORT_USAGE,
793            )
794            .expect("submit_send_zero_copy");
795        ring.submit().expect("submit");
796
797        let mut got_result = false;
798        let mut got_notif = false;
799        let mut bytes = 0usize;
800        let mut restituted = Vec::new();
801        while !(got_result && got_notif) {
802            let c = ring.wait_completion().expect("wait_completion");
803            assert_eq!(c.token(), tok);
804            if c.has_more() {
805                // CQE résultat : octets sans restitution, slot survit (in_flight 1).
806                bytes = c.into_result().expect("result ok").try_into().unwrap_or(0);
807                assert_eq!(ring.in_flight(), 1, "slot survit à la 1ʳᵉ complétion");
808                got_result = true;
809            } else if c.is_notif() {
810                // CQE NOTIF : restitution + libération du slot.
811                let _copied = c.zero_copy_copied();
812                restituted = c.into_zero_copy_buffer().expect("notif ok");
813                got_notif = true;
814            }
815        }
816        assert_eq!(bytes, 4096);
817        assert_eq!(restituted, data, "buffer restitué au NOTIF");
818        assert_eq!(ring.in_flight(), 0, "slot libéré au NOTIF");
819
820        // Draine les données côté pair (évite un blocage au Drop).
821        let mut sink = vec![0u8; 4096];
822        let _ = server.read(&mut sink);
823        drop(client);
824    }
825
826    #[test]
827    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
828    fn sendmsg_zero_copy_multi_buffer_restitutes_all() {
829        // ADR-032 : `sendmsg_zc` multi-buffers ⇒ `into_zero_copy_buffers` rend
830        // **tous** les buffers, intacts et dans l'ordre (jamais le premier seul).
831        let (client, mut server) = tcp_pair();
832        let mut ring = ring8();
833        let original = vec![vec![1u8; 1024], vec![2u8; 512], vec![3u8; 800]];
834        let send = OwnedSendMessage {
835            buffers: original.clone(),
836            address: None,
837            fds: Vec::new(),
838            flags: MessageFlags::empty(),
839        };
840        let tok = ring
841            .submit_send_message_zero_copy(sfd(&client), send, ZeroCopyFlags::empty())
842            .expect("submit_send_message_zero_copy");
843        ring.submit().expect("submit");
844        let mut seen_result = false;
845        let mut seen_notif = false;
846        let mut restituted = Vec::new();
847        while !(seen_result && seen_notif) {
848            let c = ring.wait_completion().expect("wait");
849            assert_eq!(c.token(), tok);
850            if c.has_more() {
851                let _ = c.into_result().expect("result");
852                assert_eq!(ring.in_flight(), 1, "slot survit à la 1ʳᵉ complétion");
853                seen_result = true;
854            } else if c.is_notif() {
855                restituted = c.into_zero_copy_buffers().expect("notif");
856                seen_notif = true;
857            }
858        }
859        assert_eq!(ring.in_flight(), 0);
860        assert_eq!(
861            restituted, original,
862            "TOUS les buffers restitués, intacts et dans l'ordre"
863        );
864        let total: usize = original.iter().map(Vec::len).sum();
865        let mut sink = vec![0u8; total];
866        let _ = server.read(&mut sink);
867        drop(client);
868    }
869
870    #[test]
871    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
872    fn send_zc_on_unsupported_socket_keeps_buffer_until_quiescent() {
873        // AF_UNIX ne supporte pas le zero-copy. Selon le kernel, l'échec se
874        // manifeste soit en CQE résultat `res<0` **avec** `F_MORE` + NOTIF (le
875        // buffer reste détenu jusqu'au NOTIF qui le restitue), soit en CQE unique
876        // sans `F_MORE` (échec précoce, buffer libéré via l'erreur). Les deux
877        // chemins sont **sûrs** (aucune fuite, aucune panique) ; on draine
878        // jusqu'à quiescence en exerçant les accesseurs.
879        let (a, _b) = pair();
880        let mut ring = ring8();
881        let tok = ring
882            .submit_send_zero_copy(
883                sfd(&a),
884                vec![1u8; 32],
885                MessageFlags::empty(),
886                ZeroCopyFlags::empty(),
887            )
888            .expect("submit_send_zero_copy");
889        ring.submit().expect("submit");
890        let mut restituted = Vec::new();
891        while ring.in_flight() > 0 {
892            let c = ring.wait_completion().expect("wait");
893            assert_eq!(c.token(), tok);
894            if c.is_notif() {
895                restituted = c.into_zero_copy_buffer().expect("notif restitue");
896            } else if c.has_more() {
897                // CQE résultat (peut porter une erreur) : pas de restitution.
898                let _ = c.into_result();
899            } else {
900                // CQE unique (échec précoce) : l'accesseur propage l'erreur.
901                restituted = c.into_zero_copy_buffer().unwrap_or_default();
902            }
903        }
904        assert!(
905            restituted.is_empty() || restituted.len() == 32,
906            "buffer restitué (NOTIF) ou libéré (échec précoce) — jamais corrompu"
907        );
908        assert_eq!(ring.in_flight(), 0);
909    }
910
911    // ── Back-pressure EBUSY ───────────────────────────────────────────────
912
913    #[test]
914    #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
915    fn slab_full_returns_ebusy() {
916        use crate::io_uring::IoUringBuilder;
917        let (a, _b) = pair();
918        let mut ring = IoUringBuilder::new(NonZeroU32::new(8).expect("8"))
919            .max_inflight(NonZeroU32::new(1).expect("1"))
920            .build()
921            .expect("ring slab=1");
922        let _t = ring
923            .submit_send(sfd(&a), b"x".to_vec(), MessageFlags::empty())
924            .expect("1ʳᵉ op");
925        assert_eq!(
926            ring.submit_send(sfd(&a), b"y".to_vec(), MessageFlags::empty()),
927            Err(Errno::EBUSY)
928        );
929        ring.submit_and_wait(1).expect("submit");
930        let _ = ring.wait_completion().expect("complétion");
931    }
932
933    proptest::proptest! {
934        /// `send`/`recv` : octets transférés ≤ taille du buffer, et restitution
935        /// **fidèle** du contenu (relu à l'identique).
936        #[test]
937        #[cfg_attr(miri, ignore = "syscall io_uring non supporté par Miri")]
938        fn send_recv_bounded_and_faithful(
939            data in proptest::collection::vec(proptest::prelude::any::<u8>(), 0..400)
940        ) {
941            let (a, b) = pair();
942            let mut ring = ring8();
943            let len = data.len();
944            let tok = ring
945                .submit_send(sfd(&a), data.clone(), MessageFlags::empty())
946                .expect("submit_send");
947            ring.submit_and_wait(1).expect("submit");
948            let (_buf, sent) = ring
949                .wait_completion()
950                .expect("complétion")
951                .into_buffer_result()
952                .expect("send ok");
953            let _ = tok;
954            proptest::prop_assert!(sent <= len);
955            if len == 0 {
956                return Ok(()); // rien à relire
957            }
958            proptest::prop_assert_eq!(sent, len);
959            let tok = ring
960                .submit_receive(sfd(&b), vec![0u8; len], MessageFlags::empty())
961                .expect("submit_receive");
962            ring.submit_and_wait(1).expect("submit");
963            let (buf, recv) = ring
964                .wait_completion()
965                .expect("complétion")
966                .into_buffer_result()
967                .expect("recv ok");
968            let _ = tok;
969            proptest::prop_assert!(recv <= len);
970            proptest::prop_assert_eq!(&buf[..recv], &data[..recv]);
971        }
972    }
973
974    // ── Accesseurs : bras défensifs + directions d'erreur (purs, Miri) ────
975
976    fn cqe(res: i32, raw_flags: u32, payload: Option<OwnedOp>) -> Completion {
977        Completion::for_test(SubmissionToken::new(0, 0), res, raw_flags, payload)
978    }
979
980    #[test]
981    fn fd_accessors_propagate_errors() {
982        assert_eq!(cqe(-9, 0, None).into_socket_fd().err(), Some(Errno::EBADF));
983        assert_eq!(cqe(-9, 0, None).accepted_fd().err(), Some(Errno::EBADF));
984        assert_eq!(
985            cqe(-22, 0, None).into_accept_result().err(),
986            Some(Errno::EINVAL)
987        );
988        assert_eq!(
989            cqe(-11, 0, None).into_receive_message_result().err(),
990            Some(Errno::EAGAIN)
991        );
992    }
993
994    #[test]
995    #[cfg_attr(miri, ignore = "matérialise/ferme de vrais FD (close) — hors Miri")]
996    fn accept_result_decodes_or_defaults() {
997        // Payload Accept avec adresse Unix path décodable.
998        let (storage, len) = socket_addr_to_raw(&SocketAddr::Unix(UnixSocketAddr::Path(
999            CString::new("/tmp/x").expect("nul"),
1000        )));
1001        // res ≥ 0 : un vrai FD (stdin dupliqué) que l'accesseur enveloppera en
1002        // `OwnedFd` (`from_raw_fd`) puis qu'on fermera — pas d'`unsafe` ici.
1003        let (fd, addr) = cqe(
1004            dup_stdin(),
1005            0,
1006            Some(OwnedOp::Accept {
1007                addr: Box::new(storage),
1008                addrlen: Box::new(len),
1009            }),
1010        )
1011        .into_accept_result()
1012        .expect("ok");
1013        assert!(matches!(addr, SocketAddr::Unix(UnixSocketAddr::Path(_))));
1014        drop(fd);
1015
1016        // addrlen = 0 ⇒ fallback Unnamed (bras défensif). FD = nouveau dup.
1017        let (fd2, addr2) = cqe(
1018            dup_stdin(),
1019            0,
1020            Some(OwnedOp::Accept {
1021                addr: Box::new(RawSockaddrStorage {
1022                    bytes: [0u8; MAX_SOCKADDR_LEN],
1023                }),
1024                addrlen: Box::new(0),
1025            }),
1026        )
1027        .into_accept_result()
1028        .expect("ok");
1029        assert!(matches!(addr2, SocketAddr::Unix(UnixSocketAddr::Unnamed)));
1030        drop(fd2);
1031
1032        // Payload absent ⇒ fallback Unnamed (bras `_`).
1033        let (fd3, addr3) = cqe(dup_stdin(), 0, None).into_accept_result().expect("ok");
1034        assert!(matches!(addr3, SocketAddr::Unix(UnixSocketAddr::Unnamed)));
1035        drop(fd3);
1036    }
1037
1038    /// Duplique un FD réel valide (stdin) sans dépendance externe, pour fournir
1039    /// un `res ≥ 0` aux tests d'accesseurs qui matérialisent un `OwnedFd`.
1040    fn dup_stdin() -> i32 {
1041        // Pur std : on duplique le fd 0 via l'API standard puis on extrait le
1042        // RawFd brut. Traits std qualifiés explicitement (l'`AsFd`/`IntoRawFd`
1043        // d'Air ne s'appliquent pas à `Stdin`/std `OwnedFd`).
1044        let cloned = std::os::fd::AsFd::as_fd(&std::io::stdin())
1045            .try_clone_to_owned()
1046            .expect("dup stdin");
1047        std::os::fd::IntoRawFd::into_raw_fd(cloned)
1048    }
1049
1050    #[test]
1051    fn zero_copy_buffer_paths() {
1052        // NOTIF : restitution inconditionnelle (Bytes).
1053        let notif = cqe(
1054            0,
1055            CompletionFlags::NOTIF.bits(),
1056            Some(OwnedOp::Bytes(vec![1, 2, 3])),
1057        );
1058        assert_eq!(notif.into_zero_copy_buffer().expect("ok"), vec![1, 2, 3]);
1059
1060        // NOTIF + bit copié (REPORT_USAGE) : pas une erreur ; copied = true.
1061        let copied = cqe(
1062            i32::MIN,
1063            CompletionFlags::NOTIF.bits(),
1064            Some(OwnedOp::Bytes(vec![9])),
1065        );
1066        assert!(copied.zero_copy_copied());
1067        assert_eq!(copied.into_zero_copy_buffer().expect("ok"), vec![9]);
1068
1069        // Échec précoce : pas NOTIF, res < 0 ⇒ Err.
1070        let early = cqe(-22, 0, Some(OwnedOp::Bytes(vec![5])));
1071        assert_eq!(early.into_zero_copy_buffer().err(), Some(Errno::EINVAL));
1072
1073        // SendMsg sur l'accesseur mono : concaténation sans perte d'octet (ADR-032).
1074        let send_payload =
1075            OwnedOp::SendMsg(Box::new(make_send_state(vec![vec![4u8; 2], vec![5u8]])));
1076        let notif_msg = cqe(0, CompletionFlags::NOTIF.bits(), Some(send_payload));
1077        assert_eq!(
1078            notif_msg.into_zero_copy_buffer().expect("ok"),
1079            vec![4u8, 4, 5],
1080            "concaténation : aucun octet discardé"
1081        );
1082
1083        // zero_copy_copied = false hors NOTIF.
1084        assert!(!cqe(i32::MIN, 0, None).zero_copy_copied());
1085
1086        // Mauvais payload sur NOTIF ⇒ buffer vide (bras `_`).
1087        let empty = cqe(0, CompletionFlags::NOTIF.bits(), None);
1088        assert!(empty.into_zero_copy_buffer().expect("ok").is_empty());
1089    }
1090
1091    #[test]
1092    fn into_zero_copy_buffers_restitutes_all_paths() {
1093        // sendmsg_zc multi-buffers : TOUS les buffers, dans l'ordre (ADR-032).
1094        let multi = vec![vec![1u8, 2], vec![3], vec![4u8, 5, 6]];
1095        let payload = OwnedOp::SendMsg(Box::new(make_send_state(multi.clone())));
1096        let notif = cqe(0, CompletionFlags::NOTIF.bits(), Some(payload));
1097        assert_eq!(notif.into_zero_copy_buffers().expect("ok"), multi);
1098
1099        // send_zc mono-buffer : enveloppé dans un Vec à un élément.
1100        let single = cqe(
1101            0,
1102            CompletionFlags::NOTIF.bits(),
1103            Some(OwnedOp::Bytes(vec![7, 8])),
1104        );
1105        assert_eq!(
1106            single.into_zero_copy_buffers().expect("ok"),
1107            vec![vec![7, 8]]
1108        );
1109
1110        // Échec précoce (pas NOTIF, res < 0) ⇒ Err ; buffers libérés.
1111        let early = cqe(
1112            -22,
1113            0,
1114            Some(OwnedOp::SendMsg(Box::new(make_send_state(vec![vec![9u8]])))),
1115        );
1116        assert_eq!(early.into_zero_copy_buffers().err(), Some(Errno::EINVAL));
1117
1118        // Mauvais payload sur NOTIF ⇒ vecteur vide (bras `_`).
1119        let empty = cqe(0, CompletionFlags::NOTIF.bits(), None);
1120        assert!(empty.into_zero_copy_buffers().expect("ok").is_empty());
1121    }
1122
1123    #[test]
1124    #[cfg_attr(miri, ignore = "matérialise/ferme de vrais FD (close) — hors Miri")]
1125    fn receive_message_result_defensive_and_truncation() {
1126        // Payload absent ⇒ résultat vide (bras `else`).
1127        let (msg, meta) = cqe(3, 0, None).into_receive_message_result().expect("ok");
1128        assert!(msg.buffers.is_empty());
1129        assert_eq!(meta.payloadlen, 3);
1130        assert!(meta.fds.is_empty() && meta.address.is_none());
1131
1132        // RecvMsg avec un cmsg SCM_RIGHTS valide construit à la main + adresse.
1133        let mut control = vec![0u8; 24];
1134        control[0..8].copy_from_slice(&20u64.to_ne_bytes()); // cmsg_len = 16 + 4
1135        control[8..12].copy_from_slice(&1i32.to_ne_bytes()); // SOL_SOCKET
1136        control[12..16].copy_from_slice(&1i32.to_ne_bytes()); // SCM_RIGHTS
1137        let dup = dup_stdin();
1138        control[16..20].copy_from_slice(&dup.to_ne_bytes());
1139        let (name, namelen) =
1140            socket_addr_to_raw(&SocketAddr::Ipv4(air_sys_types::net::Ipv4SocketAddr {
1141                address: [127, 0, 0, 1],
1142                port: 8080,
1143            }));
1144        let state = RecvMsgState {
1145            msghdr: Box::new(Msghdr {
1146                msg_namelen: namelen,
1147                msg_controllen: 24,
1148                msg_flags: MessageFlags::empty().bits(),
1149                ..Msghdr::default()
1150            }),
1151            iovecs: Vec::new().into_boxed_slice(),
1152            buffers: vec![vec![1u8, 2, 3]],
1153            name: Box::new(name),
1154            control: control.into_boxed_slice(),
1155        };
1156        let (msg, meta) = cqe(3, 0, Some(OwnedOp::RecvMsg(Box::new(state))))
1157            .into_receive_message_result()
1158            .expect("ok");
1159        assert_eq!(msg.buffers[0], vec![1u8, 2, 3]);
1160        assert_eq!(meta.fds.len(), 1, "FD extrait du cmsg");
1161        assert!(matches!(meta.address, Some(SocketAddr::Ipv4(_))));
1162        assert_eq!(meta.controllen, 24);
1163        drop(meta);
1164    }
1165
1166    // Helpers purs pour construire un SendMsgState de test sans kernel.
1167    fn make_send_state(buffers: Vec<Vec<u8>>) -> SendMsgState {
1168        let mut buffers = buffers;
1169        let iovecs: Box<[Iovec]> = buffers
1170            .iter_mut()
1171            .map(|b| Iovec {
1172                iov_base: b.as_mut_ptr(),
1173                iov_len: b.len(),
1174            })
1175            .collect();
1176        SendMsgState {
1177            msghdr: Box::new(Msghdr::default()),
1178            iovecs,
1179            buffers,
1180            addr: None,
1181            control: Vec::new().into_boxed_slice(),
1182            fds: Vec::new(),
1183        }
1184    }
1185
1186    use super::super::CompletionFlags;
1187
1188    // ── Ownership zero-copy à deux complétions (Miri, pur) ────────────────
1189
1190    #[test]
1191    fn ownership_zero_copy_two_phase_via_slab() {
1192        use super::super::slab::InflightSlab;
1193        let mut slab = InflightSlab::with_capacity(NonZeroU32::new(2).expect("2"));
1194        let token = slab
1195            .reserve(Some(OwnedOp::Bytes(vec![5u8; 16])))
1196            .expect("réservation");
1197        // 1ʳᵉ complétion (résultat, has_more=true) : slot survit, pas de restitution.
1198        let outcome = slab.complete(token.to_user_data(), true);
1199        assert!(matches!(outcome, super::super::slab::SlotOutcome::More));
1200        assert_eq!(slab.in_flight(), 1, "buffer toujours détenu (vivant)");
1201        // 2ᵉ complétion (NOTIF, has_more=false) : restitution + libération.
1202        let outcome = slab.complete(token.to_user_data(), false);
1203        let buffer = match outcome {
1204            super::super::slab::SlotOutcome::Final {
1205                payload: Some(OwnedOp::Bytes(b)),
1206            } => b,
1207            _ => Vec::new(),
1208        };
1209        assert_eq!(buffer, vec![5u8; 16]);
1210        assert_eq!(slab.in_flight(), 0);
1211    }
1212
1213    #[test]
1214    fn ownership_zero_copy_multi_buffer_two_phase_via_slab() {
1215        // ADR-032 + cycle 2 complétions, multi-buffers (sendmsg_zc) : aucun
1216        // buffer libéré/lu avant le NOTIF, et **tous** restitués au NOTIF (Miri).
1217        use super::super::slab::{InflightSlab, SlotOutcome};
1218        let original = vec![vec![10u8, 11], vec![12], vec![13u8, 14, 15]];
1219        let mut slab = InflightSlab::with_capacity(NonZeroU32::new(2).expect("2"));
1220        let token = slab
1221            .reserve(Some(OwnedOp::SendMsg(Box::new(make_send_state(
1222                original.clone(),
1223            )))))
1224            .expect("réservation");
1225        // Résultat (has_more) : slot vivant, aucun buffer rendu.
1226        assert!(matches!(
1227            slab.complete(token.to_user_data(), true),
1228            SlotOutcome::More
1229        ));
1230        assert_eq!(slab.in_flight(), 1, "tous les buffers toujours détenus");
1231        // NOTIF (final) : restitution intégrale via le payload.
1232        let buffers = match slab.complete(token.to_user_data(), false) {
1233            SlotOutcome::Final {
1234                payload: Some(OwnedOp::SendMsg(state)),
1235            } => state.buffers,
1236            _ => Vec::new(),
1237        };
1238        assert_eq!(buffers, original, "tous les buffers, intacts, dans l'ordre");
1239        assert_eq!(slab.in_flight(), 0);
1240    }
1241
1242    #[test]
1243    fn ownership_send_state_round_trips_under_miri() {
1244        // Construit un SendMsgState complet (msghdr pointant vers iovec/control/
1245        // addr indépendants) puis le drop : aucun double-free / UAF (Miri).
1246        let request = OwnedSendMessage {
1247            buffers: vec![vec![1u8, 2, 3], vec![4, 5]],
1248            address: Some(SocketAddr::Ipv4(air_sys_types::net::Ipv4SocketAddr {
1249                address: [127, 0, 0, 1],
1250                port: 80,
1251            })),
1252            fds: Vec::new(),
1253            flags: MessageFlags::empty(),
1254        };
1255        let (state, hdr_ptr) = build_send_state(request, MessageFlags::NOSIGNAL.bits());
1256        assert_ne!(hdr_ptr, 0);
1257        // Le msghdr pointe vers les iovec parqués (adresses stables).
1258        assert_eq!(state.iovecs.len(), 2);
1259        let payload = OwnedOp::SendMsg(state);
1260        drop(payload);
1261    }
1262}
air_sys_syscall/io_uring/net_ops.rs

air_sys_syscall/io_uring/
net_ops.rs
