feat: io_uring fixed-buffer send (write_fixed) (#284) (#497)

* feat: io_uring fixed-buffer send (write_fixed) (#284)

Completes the fixed recv/send primitive pair: `send_fixed` writes a reply from a
REGISTERED buffer via write_fixed_all, the write-side analog of recv_fixed. New
in ironcache-runtime::fixed_datapath.

It stages `data` into a checked-out registered buffer (copy_nonoverlapping into
the buffer's stable_mut_ptr + set_init, in one small documented unsafe block --
data.len() is guarded <= the buffer capacity, distinct allocations, exactly
data.len() bytes written then marked initialized, so write_fixed_all reads only
those bytes) and writes it without a per-write pin.

Its fall-back contract MIRRORS recv_fixed's for one clean caller branch: it
returns None -- "use the owned-buffer send" -- when the reply does NOT fit one
buffer (> ring.buf_size(), a rare large bulk value) OR the slab is drained. The
single-buffer size decision stays with this primitive; the serve-loop wiring
picks fixed-vs-owned send off the None.

Validated on a real ring (local Linux container kernel 6.8 + the CI io_uring
datapath job): a full fixed recv+send round-trip (request read via read_fixed,
reply written via write_fixed_all, client observes it byte-exact) plus the
oversized-reply -> None fall-back. 43 io_uring lib tests pass; clippy -D warnings
over --features io_uring clean (incl. the unsafe block); macOS build + fmt +
invariant lints clean; 0 dashes. No throughput claim (deferred to a pinned host).

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: ELares <zeke@butlr.io>

* fix: slice fixed-buffer send to data.len() to not leak stale bytes (#284)

Adversarial review (verified against the tokio-uring 0.5 source) caught a real
correctness + information-disclosure defect in send_fixed. `write_fixed_all(buf)`
submits `buf.bytes_init()` bytes, but tokio-uring's `set_init` is GROW-ONLY and
`FixedBufPool` RETAINS init_len across buffer reuse. So a reused buffer whose
retained init_len exceeds the reply length would send the reply FOLLOWED BY the
stale trailing bytes of a prior (longer) recv/reply through that buffer -- and
because the slab is shared across all connections on the shard, those leaked bytes
can be another client's request/reply data.

Not UB / not OOB (the stale bytes are within init_len = previously-initialized
memory); a semantic bug: `set_init(data.len())` cannot SHRINK init_len, so it does
not bound the write.

Fix: write `buf.slice(0..data.len())` so the SQE length is the slice length
(`end - begin`), independent of the buffer's retained init_len. The reply is
pinned to exactly the fresh bytes.

Regression test added (fixed_send_of_a_short_reply_after_a_longer_recv_leaks_no_
stale_bytes): a pool of ONE buffer forces send_fixed to reuse the exact buffer a
prior 20-byte recv filled, then sends a 5-byte reply. VERIFIED it FAILS on the old
`write_fixed_all(buf)` (client received "+OK\r\n" + 15 stale 'q' bytes from the
prior request) and PASSES with the slice fix (exactly 5 bytes). My earlier
round-trip test masked the bug only because its 7-byte reply happened to be LONGER
than the 6-byte prior recv (set_init grew, no stale tail).

44 io_uring lib tests pass; clippy -D warnings over --features io_uring clean;
macOS build + fmt + invariant lints clean; 0 dashes.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: ELares <zeke@butlr.io>

---------

Signed-off-by: ELares <zeke@butlr.io>
Co-authored-by: ELares <zeke@butlr.io>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

CHANGELOG.md

@@ -164,6 +164,12 @@ release.
 
 ### Added
 
+- io_uring fixed-buffer SEND (issue #284): `ironcache-runtime::fixed_datapath` gains `send_fixed`,
+  the reply side over registered buffers, completing the fixed recv/send pair. It stages the reply
+  into a checked-out registered buffer and writes it via `write_fixed_all` (no per-write pin). Its
+  fall-back contract mirrors `recv_fixed`: it returns `None` (use the owned-buffer send) when the
+  reply does not fit one buffer (a rare large bulk value) OR the slab is drained, so the caller has
+  one clean branch. Functionally validated on a real ring by a full fixed recv+send round-trip.
 - io_uring OneShotFixed datapath substrate (issue #284, IOURING_DATAPATH.md "Registered fixed-buffer
   slab"): `ironcache-runtime`'s new `fixed_datapath` module adds `FixedRing` (a per-shard slab of
   buffers registered once with the ring via tokio-uring's `FixedBufPool`) + `recv_fixed`, which reads

crates/ironcache-runtime/src/fixed_datapath.rs

@@ -19,6 +19,7 @@
 use std::io;
 
 use tokio_uring::buf::fixed::{FixedBuf, FixedBufPool};
+use tokio_uring::buf::{BoundedBuf, IoBufMut};
 use tokio_uring::net::TcpStream;
 
 /// A per-shard REGISTERED fixed-buffer group for the OneShotFixed datapath: `count` buffers of
@@ -86,9 +87,44 @@ pub async fn recv_fixed(
     Some(stream.read_fixed(buf).await)
 }
 
+/// Write `data` to `stream` from a checked-out REGISTERED buffer via `write_fixed_all`: the reply is
+/// staged into the pre-registered slab and written without a per-write pin. The contract mirrors
+/// `recv_fixed`'s fall-back shape -- `None` means "the fixed send does not apply, use the owned-buffer
+/// send" -- so the caller has one clean branch:
+/// - `None` if `data` does not fit ONE buffer (`> ring.buf_size()`, a rare large bulk reply) OR the
+///   group is drained (no buffer to stage into); either way the caller falls back to the owned send.
+/// - `Some(Ok(()))` on a fully-written fixed send; `Some(Err(e))` on a write error.
+pub async fn send_fixed(
+    stream: &TcpStream,
+    ring: &FixedRing,
+    data: &[u8],
+) -> Option<io::Result<()>> {
+    // A reply larger than one fixed buffer uses the owned send (signalled by None), not a multi-
+    // buffer fixed write -- keeps this primitive single-buffer and the size decision in the caller.
+    if data.len() > ring.buf_size() {
+        return None;
+    }
+    let mut buf = ring.checkout()?; // None = drained -> owned-send fall-back
+    // SAFETY: `data.len() <= ring.buf_size() <= capacity` (the `>` guard above + `Vec::with_capacity`
+    // never under-allocates), so `stable_mut_ptr()` is valid for `data.len()` writes. `data` and
+    // `buf` are distinct allocations (non-overlapping). We write exactly `data.len()` bytes then grow
+    // the init length to cover them, so the `0..data.len()` slice below has an INITIALIZED range.
+    unsafe {
+        std::ptr::copy_nonoverlapping(data.as_ptr(), buf.stable_mut_ptr(), data.len());
+        buf.set_init(data.len());
+    }
+    // Write EXACTLY `data.len()` bytes by SLICING to `0..data.len()`: the SQE length is the slice
+    // length, NOT the buffer's `bytes_init()`. tokio-uring's `set_init` is GROW-ONLY and the pool
+    // RETAINS `init_len` across buffer reuse, so a reused buffer can carry a LARGER init_len from a
+    // prior (longer) recv/reply; writing `bytes_init()` would then append those stale, possibly
+    // cross-connection, trailing bytes after the reply. The slice pins the write to the fresh reply.
+    let (res, _buf) = stream.write_fixed_all(buf.slice(0..data.len())).await;
+    Some(res)
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{FixedRing, recv_fixed};
+    use super::{FixedRing, recv_fixed, send_fixed};
     use crate::io_uring_rt::IoUringRuntime;
     use crate::{Runtime, tokio_rt};
     use tokio_uring::net::TcpListener;
@@ -141,6 +177,103 @@ mod tests {
         });
     }
 
+    /// The full FIXED round-trip: the server reads the request via `recv_fixed` AND writes the reply
+    /// via `send_fixed` (both over the registered slab). Also asserts the fall-back contract: a reply
+    /// larger than one buffer yields `None` (caller uses the owned send).
+    #[test]
+    fn fixed_ring_recv_and_send_round_trip_over_the_registered_slab() {
+        tokio_uring::start(async {
+            let std_listener =
+                tokio_rt::bind_reuseport_std("127.0.0.1:0".parse().unwrap()).unwrap();
+            let addr = std_listener.local_addr().unwrap();
+            let listener = TcpListener::from_std(std_listener);
+
+            let server = tokio_uring::spawn(async move {
+                let runtime = IoUringRuntime::new();
+                let (stream, _peer) = runtime.accept(&listener).await.unwrap();
+                let ring = FixedRing::register(8, 4096).unwrap();
+
+                let (res, buf) = recv_fixed(&stream, &ring).await.expect("buffer available");
+                let n = res.unwrap();
+                assert_eq!(&buf[..n], b"PING\r\n");
+                drop(buf);
+
+                // A reply that does NOT fit one buffer -> None (caller falls back to owned send).
+                let too_big = vec![b'x'; ring.buf_size() + 1];
+                assert!(
+                    send_fixed(&stream, &ring, &too_big).await.is_none(),
+                    "oversized reply must signal owned-send fall-back"
+                );
+
+                // The fitting reply is written from the registered slab.
+                send_fixed(&stream, &ring, b"+PONG\r\n")
+                    .await
+                    .expect("fits + not drained")
+                    .unwrap();
+            });
+
+            let client = IoUringRuntime::new();
+            let mut peer = client.connect(addr).await.unwrap();
+            let _ = client.send(&mut peer, b"PING\r\n".to_vec()).await.unwrap();
+            let res = client
+                .recv(&mut peer, Vec::with_capacity(16))
+                .await
+                .unwrap();
+            assert_eq!(&res.buf[..res.n], b"+PONG\r\n");
+            server.await.unwrap();
+        });
+    }
+
+    /// Regression: a SHORT reply after a LONGER recv on the SAME reused buffer must write only the
+    /// reply, never the stale trailing bytes of the prior read. tokio-uring's `set_init` is grow-only
+    /// and the pool retains `init_len` across reuse, so `write_fixed_all(buf)` would send
+    /// `bytes_init()` (= the prior 20) bytes; `send_fixed` slices to `data.len()` to prevent that
+    /// leak. A pool of ONE buffer forces `send_fixed` to reuse the exact buffer `recv_fixed` filled.
+    #[test]
+    fn fixed_send_of_a_short_reply_after_a_longer_recv_leaks_no_stale_bytes() {
+        tokio_uring::start(async {
+            let std_listener =
+                tokio_rt::bind_reuseport_std("127.0.0.1:0".parse().unwrap()).unwrap();
+            let addr = std_listener.local_addr().unwrap();
+            let listener = TcpListener::from_std(std_listener);
+
+            let server = tokio_uring::spawn(async move {
+                let runtime = IoUringRuntime::new();
+                let (stream, _peer) = runtime.accept(&listener).await.unwrap();
+                // ONE buffer: send_fixed necessarily reuses the buffer recv_fixed just filled, whose
+                // retained init_len is 20 -- the exact grow-only trap.
+                let ring = FixedRing::register(1, 4096).unwrap();
+                let (res, buf) = recv_fixed(&stream, &ring).await.expect("buffer available");
+                let n = res.unwrap();
+                assert_eq!(
+                    n, 20,
+                    "the long request set the reused buffer's init_len to 20"
+                );
+                drop(buf);
+                // A 5-byte reply through that same buffer: must be exactly 5 bytes on the wire.
+                send_fixed(&stream, &ring, b"+OK\r\n")
+                    .await
+                    .expect("fits + not drained")
+                    .unwrap();
+            });
+
+            let client = IoUringRuntime::new();
+            let mut peer = client.connect(addr).await.unwrap();
+            let _ = client.send(&mut peer, vec![b'q'; 20]).await.unwrap();
+            let res = client
+                .recv(&mut peer, Vec::with_capacity(64))
+                .await
+                .unwrap();
+            // With the bug this would be 20 bytes ("+OK\r\n" + 15 stale 'q's); the slice makes it 5.
+            assert_eq!(
+                &res.buf[..res.n],
+                b"+OK\r\n",
+                "reply must be exactly the 5 fresh bytes, no stale trailing bytes leaked"
+            );
+            server.await.unwrap();
+        });
+    }
+
     /// Draining the group yields `None` from `checkout` (the back-pressure signal): after checking
     /// out all `count` buffers, the next check-out is `None`, and returning one makes it available
     /// again.