Repository: bearcove/loona (language: rust)

Issue:
hring-bingo-ktls: /image/jpeg is corrupted over h2
https://httpbingo.org/image/jpeg is:

![image](https://user-images.githubusercontent.com/7998310/213916889-37665c82-b246-4aea-934e-a58dd58682dd.png)

But https://localhost:7007/image/jpeg (from `hring-bingo-ktls`) shows:

![image](https://user-images.githubusercontent.com/7998310/213916892-c8d94a71-5215-4bac-817b-050a13d84e8e.png)

Most probably some errors while handling the body. Haven't compared with HTTP/1.1 and/or without TLS


Discussion / hints:

Fix patch:
diff --git a/crates/hring-bingo-ktls/Cargo.toml b/crates/hring-bingo-ktls/Cargo.toml
index 2dec646f..f1d04ff6 100644
--- a/crates/hring-bingo-ktls/Cargo.toml
+++ b/crates/hring-bingo-ktls/Cargo.toml
@@ -14,3 +14,4 @@ rustls = { version = "0.20.8", features = ["secret_extraction"] }
 tokio = { version = "1.24.2", features = ["full"] }
 tokio-rustls = "0.23.4"
 http = "0.2.8"
+pretty-hex = "0.3.0"
diff --git a/crates/hring-bingo-ktls/src/main.rs b/crates/hring-bingo-ktls/src/main.rs
index 267059ee..22b134e8 100644
--- a/crates/hring-bingo-ktls/src/main.rs
+++ b/crates/hring-bingo-ktls/src/main.rs
@@ -10,8 +10,8 @@ use std::{
 
 use color_eyre::eyre;
 use hring::{
-    h1, h2, tokio_uring::net::TcpStream, Body, Encoder, ExpectResponseHeaders, Responder,
-    ResponseDone, RollMut, ServerDriver,
+    buffet::RollMut, h1, h2, tokio_uring::net::TcpStream, Body, Encoder, ExpectResponseHeaders,
+    Method, Request, Responder, ResponseDone, ServerDriver,
 };
 use http::Version;
 use rustls::ServerConfig;
@@ -31,6 +31,11 @@ async fn async_main() -> eyre::Result<()> {
         )
         .init();
 
+    if std::env::args().any(|a| a == "--get") {
+        sample_http_request().await.unwrap();
+        return Ok(());
+    }
+
     let pair = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])?;
     let crt = pair.serialize_der()?;
     let key = pair.serialize_private_key_der();
@@ -48,29 +53,117 @@ async fn async_main() -> eyre::Result<()> {
     let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
     let acceptor = Rc::new(acceptor);
 
-    let ln = TcpListener::bind("[::]:7007").await?;
-    info!("Listening on {}", ln.local_addr()?);
+    let pt_h1_ln = TcpListener::bind("[::]:7080").await?;
+    info!("Serving plaintext HTTP/1.1 on {}", pt_h1_ln.local_addr()?);
+
+    let pt_h2_ln = TcpListener::bind("[::]:7082").await?;
+    info!("Serving plaintext HTTP/2 on {}", pt_h2_ln.local_addr()?);
+
+    let tls_ln = TcpListener::bind("[::]:7443").await?;
+    info!("Serving HTTPS on {}", tls_ln.local_addr()?);
 
     let h1_conf = Rc::new(h1::ServerConf::default());
     let h2_conf = Rc::new(h2::ServerConf::default());
 
-    while let Ok((stream, remote_addr)) = ln.accept().await {
-        hring::tokio_uring::spawn({
-            let acceptor = acceptor.clone();
-            let h1_conf = h1_conf.clone();
-            let h2_conf = h2_conf.clone();
-            async move {
-                if let Err(e) = handle_conn(acceptor, stream, remote_addr, h1_conf, h2_conf).await {
-                    tracing::error!(%e, "Error handling connection");
-                }
+    let pt_h1_loop = {
+        let h1_conf = h1_conf.clone();
+
+        async move {
+            while let Ok((stream, remote_addr)) = pt_h1_ln.accept().await {
+                hring::tokio_uring::spawn({
+                    let h1_conf = h1_conf.clone();
+                    async move {
+                        if let Err(e) =
+                            handle_plaintext_conn(stream, remote_addr, Proto::H1(h1_conf)).await
+                        {
+                            tracing::error!(%e, "Error handling connection");
+                        }
+                    }
+                });
             }
-        });
+
+            Ok::<_, color_eyre::Report>(())
+        }
+    };
+
+    let pt_h2_loop = {
+        let h2_conf = h2_conf.clone();
+
+        async move {
+            while let Ok((stream, remote_addr)) = pt_h2_ln.accept().await {
+                hring::tokio_uring::spawn({
+                    let h2_conf = h2_conf.clone();
+                    async move {
+                        if let Err(e) =
+                            handle_plaintext_conn(stream, remote_addr, Proto::H2(h2_conf)).await
+                        {
+                            tracing::error!(%e, "Error handling connection");
+                        }
+                    }
+                });
+            }
+
+            Ok::<_, color_eyre::Report>(())
+        }
+    };
+
+    let tls_loop = async move {
+        while let Ok((stream, remote_addr)) = tls_ln.accept().await {
+            hring::tokio_uring::spawn({
+                let acceptor = acceptor.clone();
+                let h1_conf = h1_conf.clone();
+                let h2_conf = h2_conf.clone();
+                async move {
+                    if let Err(e) =
+                        handle_tls_conn(acceptor, stream, remote_addr, h1_conf, h2_conf).await
+                    {
+                        tracing::error!(%e, "Error handling connection");
+                    }
+                }
+            });
+        }
+
+        Ok::<_, color_eyre::Report>(())
+    };
+
+    tokio::try_join!(pt_h1_loop, pt_h2_loop, tls_loop)?;
+    Ok(())
+}
+
+enum Proto {
+    H1(Rc<h1::ServerConf>),
+    H2(Rc<h2::ServerConf>),
+}
+
+async fn handle_plaintext_conn(
+    stream: tokio::net::TcpStream,
+    remote_addr: std::net::SocketAddr,
+    proto: Proto,
+) -> Result<(), color_eyre::Report> {
+    info!("Accepted connection from {remote_addr}");
+    let buf = RollMut::alloc()?;
+
+    let fd = stream.as_raw_fd();
+    std::mem::forget(stream);
+    let stream = unsafe { TcpStream::from_raw_fd(fd) };
+
+    let driver = SDriver {};
+
+    match proto {
+        Proto::H1(h1_conf) => {
+            info!("Using HTTP/1.1");
+            hring::h1::serve(stream, h1_conf, buf, driver).await?;
+        }
+        Proto::H2(h2_conf) => {
+            info!("Using HTTP/2");
+            hring::h2::serve(stream, h2_conf, buf, Rc::new(driver)).await?;
+        }
     }
 
     Ok(())
 }
 
-async fn handle_conn(
+async fn handle_tls_conn(
     acceptor: Rc<tokio_rustls::TlsAcceptor>,
     stream: tokio::net::TcpStream,
     remote_addr: std::net::SocketAddr,
@@ -180,6 +273,7 @@ where
             match body.next_chunk().await? {
                 hring::BodyChunk::Chunk(chunk) => {
                     debug!("Client got chunk of len {}", chunk.len());
+
                     respond.write_chunk(chunk).await?;
                 }
                 hring::BodyChunk::Done { trailers } => {
@@ -191,3 +285,62 @@ where
         respond.finish_body(trailers).await
     }
 }
+
+struct SampleCDriver {}
+
+impl h1::ClientDriver for SampleCDriver {
+    type Return = ();
+
+    async fn on_informational_response(&mut self, _res: hring::Response) -> eyre::Result<()> {
+        // ignore informational responses
+
+        Ok(())
+    }
+
+    async fn on_final_response(
+        self,
+        res: hring::Response,
+        body: &mut impl Body,
+    ) -> eyre::Result<Self::Return> {
+        info!("Client got final response: {}", res.status);
+
+        loop {
+            debug!("Reading from body {body:?}");
+            match body.next_chunk().await? {
+                hring::BodyChunk::Chunk(chunk) => {
+                    debug!("Client got chunk of len {}", chunk.len());
+                }
+                hring::BodyChunk::Done { .. } => {
+                    break;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+async fn sample_http_request() -> color_eyre::Result<()> {
+    info!("Doing sample HTTP request to httpbingo");
+
+    let addr = "httpbingo.org:80"
+        .to_socket_addrs()?
+        .next()
+        .expect("http bingo should be up");
+    let transport = Rc::new(TcpStream::connect(addr).await?);
+    debug!("Connected to httpbingo");
+
+    let driver = SampleCDriver {};
+
+    let req = Request {
+        method: Method::Get,
+        uri: "http://httpbingo.org/image/jpeg".parse().unwrap(),
+        version: Version::HTTP_11,
+        headers: Default::default(),
+    };
+
+    let (transport, _) = h1::request(transport, req, &mut (), driver).await?;
+    // don't re-use transport for now
+    drop(transport);
+
+    Ok(())
+}
diff --git a/crates/hring-buffet/Cargo.toml b/crates/hring-buffet/Cargo.toml
new file mode 100644
index 00000000..0ff5307e
--- /dev/null
+++ b/crates/hring-buffet/Cargo.toml
@@ -0,0 +1,31 @@
+[package]
+name = "hring-buffet"
+version = "0.1.0"
+edition = "2021"
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/hapsoc/hring"
+documentation = "https://docs.rs/hring-buffet"
+readme = "README.md"
+description = """
+Buffer management for the `hring` crate.
+"""
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[features]
+miri = []
+
+[dependencies]
+eyre = "0.6.8"
+http = "0.2.8"
+memchr = "2.5.0"
+memmap2 = { version = "0.5.8", default-features = false }
+nom = "7.1.3"
+pretty-hex = "0.3.0"
+thiserror = { version = "1.0.38", default-features = false }
+tokio = { version = "1.25.0", features = ["sync", "macros"] }
+tokio-uring = "0.4.0"
+tracing = "0.1.37"
+
+[dev-dependencies]
+pretty_assertions = "1.3.0"
diff --git a/crates/hring/src/io/mod.rs b/crates/hring-buffet/src/io.rs
similarity index 100%
rename from crates/hring/src/io/mod.rs
rename to crates/hring-buffet/src/io.rs
diff --git a/crates/hring/src/io/chan.rs b/crates/hring-buffet/src/io/chan.rs
similarity index 99%
rename from crates/hring/src/io/chan.rs
rename to crates/hring-buffet/src/io/chan.rs
index b0e5da11..bb321ff2 100644
--- a/crates/hring/src/io/chan.rs
+++ b/crates/hring-buffet/src/io/chan.rs
@@ -184,7 +184,7 @@ impl WriteOwned for ChanWrite {
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, not(feature = "miri")))]
 mod tests {
     use std::{cell::RefCell, rc::Rc};
 
diff --git a/crates/hring/src/buffet/mod.rs b/crates/hring-buffet/src/lib.rs
similarity index 93%
rename from crates/hring/src/buffet/mod.rs
rename to crates/hring-buffet/src/lib.rs
index e4c416f7..53996218 100644
--- a/crates/hring/src/buffet/mod.rs
+++ b/crates/hring-buffet/src/lib.rs
@@ -1,9 +1,16 @@
+#![allow(incomplete_features)]
+#![feature(thread_local)]
+#![feature(async_fn_in_trait)]
+
 mod roll;
 pub use roll::*;
 
 mod piece;
 pub use piece::*;
 
+mod io;
+pub use io::*;
+
 use std::{
     cell::{RefCell, RefMut},
     collections::VecDeque,
@@ -15,8 +22,14 @@ use memmap2::MmapMut;
 
 pub const BUF_SIZE: u16 = 4096;
 
+#[cfg(not(feature = "miri"))]
+pub const NUM_BUF: u32 = 64 * 1024;
+
+#[cfg(feature = "miri")]
+pub const NUM_BUF: u32 = 64;
+
 #[thread_local]
-static BUF_POOL: BufPool = BufPool::new_empty(BUF_SIZE, 64 * 1024);
+static BUF_POOL: BufPool = BufPool::new_empty(BUF_SIZE, NUM_BUF);
 
 thread_local! {
     static BUF_POOL_DESTRUCTOR: RefCell<Option<MmapMut>> = RefCell::new(None);
@@ -103,11 +116,24 @@ impl BufPool {
         let mut inner = self.inner.borrow_mut();
         if inner.is_none() {
             let len = self.num_buf as usize * self.buf_size as usize;
-            let mut map = memmap2::MmapOptions::new().len(len).map_anon()?;
-            let ptr = map.as_mut_ptr();
-            BUF_POOL_DESTRUCTOR.with(|destructor| {
-                *destructor.borrow_mut() = Some(map);
-            });
+
+            let ptr: *mut u8;
+
+            #[cfg(feature = "miri")]
+            {
+                let mut map = vec![0; len];
+                ptr = map.as_mut_ptr();
+                std::mem::forget(map);
+            }
+
+            #[cfg(not(feature = "miri"))]
+            {
+                let mut map = memmap2::MmapOptions::new().len(len).map_anon()?;
+                ptr = map.as_mut_ptr();
+                BUF_POOL_DESTRUCTOR.with(|destructor| {
+                    *destructor.borrow_mut() = Some(map);
+                });
+            }
 
             let mut free = VecDeque::with_capacity(self.num_buf as usize);
             for i in 0..self.num_buf {
@@ -419,12 +445,9 @@ impl Drop for Buf {
 
 #[cfg(test)]
 mod tests {
+    use crate::{Buf, BufMut, BUF_POOL};
     use std::rc::Rc;
 
-    use crate::buffet::{Buf, BUF_POOL};
-
-    use super::BufMut;
-
     #[test]
     fn size_test() {
         assert_eq!(8, std::mem::size_of::<BufMut>());
diff --git a/crates/hring/src/buffet/piece.rs b/crates/hring-buffet/src/piece.rs
similarity index 100%
rename from crates/hring/src/buffet/piece.rs
rename to crates/hring-buffet/src/piece.rs
diff --git a/crates/hring/src/buffet/roll.rs b/crates/hring-buffet/src/roll.rs
similarity index 96%
rename from crates/hring/src/buffet/roll.rs
rename to crates/hring-buffet/src/roll.rs
index 46cc5f0a..c5f5b1ba 100644
--- a/crates/hring/src/buffet/roll.rs
+++ b/crates/hring-buffet/src/roll.rs
@@ -29,6 +29,24 @@ enum StorageMut {
     Box(BoxStorage),
 }
 
+impl Debug for StorageMut {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Buf(bm) => f
+                .debug_struct("Buf")
+                .field("index", &bm.index)
+                .field("off", &bm.off)
+                .field("len", &bm.len)
+                .finish(),
+            Self::Box(bs) => f
+                .debug_struct("Box")
+                .field("buf", &bs.buf)
+                .field("off", &bs.off)
+                .finish(),
+        }
+    }
+}
+
 impl StorageMut {
     #[inline(always)]
     fn cap(&self) -> usize {
@@ -224,8 +242,9 @@ impl RollMut {
     ) -> (std::io::Result<usize>, Self) {
         let read_cap = std::cmp::min(limit, self.cap());
         assert!(read_cap > 0, "refusing to do empty read");
-
         let read_off = self.len;
+
+        tracing::trace!(%read_off, %read_cap, storage = ?self.storage, len = %self.len, "read_into in progress...");
         let read_into = ReadInto {
             buf: self,
             off: read_off,
@@ -233,6 +252,7 @@ impl RollMut {
             init: 0,
         };
         let (res, mut read_into) = r.read(read_into).await;
+        tracing::trace!(init = %read_into.init, "read_into done!");
         read_into.buf.len += read_into.init;
         (res, read_into.buf)
     }
@@ -318,7 +338,9 @@ impl RollMut {
                 assert_eq!(ours.index, theirs.index, "roll must be from same buffer");
                 assert!(theirs.off >= ours.off, "roll must start within buffer");
                 let skipped = theirs.off - ours.off;
+                tracing::trace!(our_index = %ours.index, their_index = %theirs.index, our_off = %ours.off, their_off = %theirs.off, %skipped, "RollMut::keep");
                 self.len -= skipped as u32;
+                ours.len -= skipped;
                 ours.off = theirs.off;
             }
             (StorageMut::Box(ours), RollInner::Box(theirs)) => {
@@ -871,7 +893,7 @@ impl RollStr {
 #[cfg(test)]
 mod tests {
     use nom::IResult;
-    use tokio_uring::net::{TcpListener, TcpStream};
+    use tracing::trace;
 
     use crate::{ChanRead, Roll, RollMut, BUF_SIZE};
 
@@ -1004,6 +1026,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(not(feature = "miri"))]
     fn test_roll_readfrom_start() {
         tokio_uring::start(async move {
             let mut rm = RollMut::alloc().unwrap();
@@ -1134,7 +1157,10 @@ mod tests {
     }
 
     #[test]
+    #[cfg(not(feature = "miri"))]
     fn test_roll_iobuf() {
+        use tokio_uring::net::{TcpListener, TcpStream};
+
         async fn test_roll_iobuf_inner(mut rm: RollMut) -> eyre::Result<()> {
             rm.put(b"hello").unwrap();
             let roll = rm.take_all();
@@ -1228,6 +1254,11 @@ mod tests {
         let mut pending = &input[..];
 
         loop {
+            if buf.cap() == 0 {
+                trace!("buf had zero cap, growing");
+                buf.grow()
+            }
+
             let (rest, version) = match parse(buf.filled()) {
                 Ok(t) => t,
                 Err(e) => {
diff --git a/crates/hring/Cargo.toml b/crates/hring/Cargo.toml
index afb3b7ac..8a164b2f 100644
--- a/crates/hring/Cargo.toml
+++ b/crates/hring/Cargo.toml
@@ -15,10 +15,10 @@ byteorder = "1.4.3"
 enum-repr = "0.2.6"
 enumflags2 = "0.7.5"
 eyre = { version = "0.6.8", default-features = false }
+hring-buffet = { version = "0.1.0", path = "../hring-buffet" }
 hring-hpack = { path = "../hring-hpack" }
 http = "0.2.8"
 memchr = "2.5.0"
-memmap2 = { version = "0.5.8", default-features = false }
 nom = { version = "7.1.3", default-features = false }
 pretty-hex = { version = "0.3.0", default-features = false }
 smallvec = { version = "1.10.0", default-features = false, features = ["const_generics", "const_new", "union"] }
diff --git a/crates/hring/src/h1/body.rs b/crates/hring/src/h1/body.rs
index 6b913c3c..a8261953 100644
--- a/crates/hring/src/h1/body.rs
+++ b/crates/hring/src/h1/body.rs
@@ -4,8 +4,9 @@ use tracing::debug;
 
 use crate::{
     util::{read_and_parse, write_all_list},
-    Body, BodyChunk, BodyErrorReason, Piece, PieceList, ReadOwned, RollMut, WriteOwned,
+    Body, BodyChunk, BodyErrorReason,
 };
+use hring_buffet::{Piece, PieceList, ReadOwned, RollMut, WriteOwned};
 
 /// An HTTP/1.1 body, either chunked or content-length.
 pub(crate) struct H1Body<T> {
diff --git a/crates/hring/src/h1/client.rs b/crates/hring/src/h1/client.rs
index fde18331..df820aa3 100644
--- a/crates/hring/src/h1/client.rs
+++ b/crates/hring/src/h1/client.rs
@@ -5,12 +5,11 @@ use http::header;
 use tracing::debug;
 
 use crate::{
-    buffet::PieceList,
-    io::ReadWriteOwned,
     types::Request,
     util::{read_and_parse, write_all_list},
-    Body, HeadersExt, Response, RollMut,
+    Body, HeadersExt, Response,
 };
+use hring_buffet::{PieceList, ReadWriteOwned, RollMut};
 
 use super::{
     body::{write_h1_body, BodyWriteMode, H1Body, H1BodyKind},
diff --git a/crates/hring/src/h1/encode.rs b/crates/hring/src/h1/encode.rs
index fd38e8e5..5db88644 100644
--- a/crates/hring/src/h1/encode.rs
+++ b/crates/hring/src/h1/encode.rs
@@ -4,11 +4,11 @@ use eyre::Context;
 use http::{StatusCode, Version};
 
 use crate::{
-    buffet::PieceList,
     types::{Headers, Request, Response},
     util::write_all_list,
-    Encoder, Piece, WriteOwned,
+    Encoder,
 };
+use hring_buffet::{Piece, PieceList, WriteOwned};
 
 use super::body::{write_h1_body_chunk, write_h1_body_end, BodyWriteMode};
 
diff --git a/crates/hring/src/h1/parse.rs b/crates/hring/src/h1/parse.rs
index f9efb257..c38db865 100644
--- a/crates/hring/src/h1/parse.rs
+++ b/crates/hring/src/h1/parse.rs
@@ -13,8 +13,9 @@ use nom::{
 
 use crate::{
     types::{Headers, Request, Response},
-    Method, PieceStr, Roll, RollStr,
+    Method,
 };
+use hring_buffet::{PieceStr, Roll, RollStr};
 
 const CRLF: &[u8] = b"\r\n";
 
diff --git a/crates/hring/src/h1/server.rs b/crates/hring/src/h1/server.rs
index 4d032dcd..97706912 100644
--- a/crates/hring/src/h1/server.rs
+++ b/crates/hring/src/h1/server.rs
@@ -6,8 +6,9 @@ use tracing::debug;
 use crate::{
     h1::body::{H1Body, H1BodyKind},
     util::{read_and_parse, SemanticError},
-    ExpectResponseHeaders, HeadersExt, ReadWriteOwned, Responder, RollMut, ServerDriver,
+    ExpectResponseHeaders, HeadersExt, Responder, ServerDriver,
 };
+use hring_buffet::{ReadWriteOwned, RollMut};
 
 use super::encode::H1Encoder;
 
diff --git a/crates/hring/src/h2/body.rs b/crates/hring/src/h2/body.rs
index 698b11ab..a51246f8 100644
--- a/crates/hring/src/h2/body.rs
+++ b/crates/hring/src/h2/body.rs
@@ -1,6 +1,7 @@
 use tokio::sync::mpsc;
 
-use crate::{Body, BodyChunk, Piece};
+use crate::{Body, BodyChunk};
+use hring_buffet::Piece;
 
 #[derive(Debug)]
 pub(crate) struct H2Body {
diff --git a/crates/hring/src/h2/encode.rs b/crates/hring/src/h2/encode.rs
index 6f64fa04..c2566733 100644
--- a/crates/hring/src/h2/encode.rs
+++ b/crates/hring/src/h2/encode.rs
@@ -4,7 +4,8 @@ use http::{StatusCode, Version};
 use tokio::sync::mpsc;
 use tracing::{debug, warn};
 
-use crate::{h1::body::BodyWriteMode, Encoder, Piece, Response, Roll};
+use crate::{h1::body::BodyWriteMode, Encoder, Response};
+use hring_buffet::{Piece, Roll};
 
 use super::parse::{KnownErrorCode, StreamId};
 
@@ -86,7 +87,7 @@ impl Encoder for H2Encoder {
     // TODO: BodyWriteMode is not relevant for h2
     async fn write_body_chunk(
         &mut self,
-        chunk: crate::Piece,
+        chunk: hring_buffet::Piece,
         _mode: BodyWriteMode,
     ) -> eyre::Result<()> {
         debug!("H2Encoder::write_body_chunk");
diff --git a/crates/hring/src/h2/parse.rs b/crates/hring/src/h2/parse.rs
index 86868c8f..55ed05e0 100644
--- a/crates/hring/src/h2/parse.rs
+++ b/crates/hring/src/h2/parse.rs
@@ -14,7 +14,7 @@ use nom::{
     IResult,
 };
 
-use crate::{Roll, WriteOwned};
+use hring_buffet::{Roll, WriteOwned};
 
 /// This is sent by h2 clients after negotiating over ALPN, or when doing h2c.
 pub const PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
diff --git a/crates/hring/src/h2/server.rs b/crates/hring/src/h2/server.rs
index 2064a6f0..ea3bb311 100644
--- a/crates/hring/src/h2/server.rs
+++ b/crates/hring/src/h2/server.rs
@@ -21,9 +21,9 @@ use crate::{
         },
     },
     util::read_and_parse,
-    ExpectResponseHeaders, Headers, Method, Piece, PieceStr, ReadWriteOwned, Request, Responder,
-    Roll, RollMut, ServerDriver,
+    ExpectResponseHeaders, Headers, Method, Request, Responder, ServerDriver,
 };
+use hring_buffet::{Piece, PieceStr, ReadWriteOwned, Roll, RollMut};
 
 /// HTTP/2 server configuration
 pub struct ServerConf {
@@ -508,6 +508,8 @@ async fn h2_write_loop(
 ) -> eyre::Result<()> {
     let mut hpack_enc = hring_hpack::Encoder::new();
 
+    let mut index = 0;
+
     while let Some(ev) = ev_rx.recv().await {
         trace!("h2_write_loop: received H2 event");
         match ev {
@@ -546,12 +548,16 @@ async fn h2_write_loop(
                         let (res, _headers_encoded) = transport.write_all(headers_encoded).await;
                         res?;
                     }
-                    H2EventPayload::BodyChunk(piece) => {
+                    H2EventPayload::BodyChunk(chunk) => {
+                        let path = format!("/tmp/chunk-write-{index:06}.bin");
+                        std::fs::write(path, chunk.as_ref()).unwrap();
+                        index += 1;
+
                         let flags = BitFlags::<DataFlags>::default();
                         let frame = Frame::new(FrameType::Data(flags), ev.stream_id)
-                            .with_len(piece.len().try_into().unwrap());
+                            .with_len(chunk.len().try_into().unwrap());
                         frame.write(transport.as_ref()).await?;
-                        let (res, _) = transport.write_all(piece).await;
+                        let (res, _) = transport.write_all(chunk).await;
                         res?;
                     }
                     H2EventPayload::BodyEnd => {
@@ -648,7 +654,6 @@ fn end_headers(
                     let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
                     if value.len() == 0 || path.replace(value).is_some() {
                         unreachable!(); // No empty path nor duplicate allowed.
-
                     }
                 }
                 b"authority" => {
diff --git a/crates/hring/src/lib.rs b/crates/hring/src/lib.rs
index 5487296b..741deaf9 100644
--- a/crates/hring/src/lib.rs
+++ b/crates/hring/src/lib.rs
@@ -1,17 +1,10 @@
 #![allow(incomplete_features)]
-#![feature(thread_local)]
 #![feature(type_alias_impl_trait)]
 #![feature(async_fn_in_trait)]
 #![feature(return_position_impl_trait_in_trait)]
 
 mod util;
 
-mod buffet;
-pub use buffet::*;
-
-mod io;
-pub use io::*;
-
 mod types;
 pub use types::*;
 
@@ -21,6 +14,8 @@ pub mod h2;
 mod responder;
 pub use responder::*;
 
+pub use hring_buffet as buffet;
+
 /// re-exported so consumers can use whatever forked version we use
 pub use http;
 
diff --git a/crates/hring/src/responder.rs b/crates/hring/src/responder.rs
index df08798e..b62c0450 100644
--- a/crates/hring/src/responder.rs
+++ b/crates/hring/src/responder.rs
@@ -1,7 +1,8 @@
 use http::header;
 use tracing::debug;
 
-use crate::{h1::body::BodyWriteMode, Body, BodyChunk, Headers, HeadersExt, Piece, Response};
+use crate::{h1::body::BodyWriteMode, Body, BodyChunk, Headers, HeadersExt, Response};
+use hring_buffet::Piece;
 
 pub trait ResponseState {}
 
diff --git a/crates/hring/src/types/headers.rs b/crates/hring/src/types/headers.rs
index 00593615..cae8cb46 100644
--- a/crates/hring/src/types/headers.rs
+++ b/crates/hring/src/types/headers.rs
@@ -2,7 +2,7 @@
 
 use http::{header, HeaderMap};
 
-use crate::Piece;
+use hring_buffet::Piece;
 
 pub type Headers = HeaderMap<Piece>;
 
diff --git a/crates/hring/src/types/method.rs b/crates/hring/src/types/method.rs
index 7ab1e9ed..69ac2de9 100644
--- a/crates/hring/src/types/method.rs
+++ b/crates/hring/src/types/method.rs
@@ -1,6 +1,6 @@
 use std::fmt;
 
-use crate::{Piece, PieceStr};
+use hring_buffet::{Piece, PieceStr};
 
 /// An HTTP method, see https://httpwg.org/specs/rfc9110.html#methods
 #[derive(Clone)]
diff --git a/crates/hring/src/types/mod.rs b/crates/hring/src/types/mod.rs
index 8d3614d0..75128804 100644
--- a/crates/hring/src/types/mod.rs
+++ b/crates/hring/src/types/mod.rs
@@ -3,7 +3,7 @@ use std::fmt::{self, Debug};
 use http::{StatusCode, Uri, Version};
 use tracing::debug;
 
-use crate::Piece;
+use hring_buffet::Piece;
 
 mod headers;
 pub use headers::*;
diff --git a/crates/hring/src/util.rs b/crates/hring/src/util.rs
index e82c639a..8025f85c 100644
--- a/crates/hring/src/util.rs
+++ b/crates/hring/src/util.rs
@@ -3,11 +3,7 @@ use nom::IResult;
 use pretty_hex::PrettyHex;
 use tracing::{debug, trace};
 
-use crate::{
-    buffet::PieceList,
-    io::{ReadOwned, WriteOwned},
-    Roll, RollMut,
-};
+use hring_buffet::{PieceList, ReadOwned, Roll, RollMut, WriteOwned};
 
 /// Returns `None` on EOF, error if partially parsed message.
 pub(crate) async fn read_and_parse<Parser, Output>(


Test patch:
diff --git a/.config/nextest.toml b/.config/nextest.toml
index a40c6c9f..2703d01a 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -1,5 +1,8 @@
 [profile.default]
 slow-timeout = { period = "5s", terminate-after = 3 }
 
+[profile.miri]
+slow-timeout = { period = "60s", terminate-after = 5 }
+
 [profile.ci]
 slow-timeout = { period = "5s", terminate-after = 6 }
\ No newline at end of file
diff --git a/crates/hring/tests/integration_test.rs b/crates/hring/tests/integration_test.rs
index 5dd581cb..a7a0b295 100644
--- a/crates/hring/tests/integration_test.rs
+++ b/crates/hring/tests/integration_test.rs
@@ -7,10 +7,10 @@ mod helpers;
 use bytes::BytesMut;
 use curl::easy::{Easy, HttpVersion, List};
 use hring::{
-    h1, h2, Body, BodyChunk, ChanRead, ChanWrite, Encoder, ExpectResponseHeaders, Headers,
-    HeadersExt, Method, Piece, ReadWritePair, Request, Responder, Response, ResponseDone, RollMut,
-    ServerDriver,
+    h1, h2, Body, BodyChunk, Encoder, ExpectResponseHeaders, Headers, HeadersExt, Method, Request,
+    Responder, Response, ResponseDone, ServerDriver,
 };
+use hring_buffet::{ChanRead, ChanWrite, Piece, ReadWritePair, RollMut};
 use http::{header, StatusCode};
 use httparse::{Status, EMPTY_HEADER};
 use pretty_assertions::assert_eq;
diff --git a/crates/hring/tests/proxy.rs b/crates/hring/tests/proxy.rs
index c439138e..15a30ce4 100644
--- a/crates/hring/tests/proxy.rs
+++ b/crates/hring/tests/proxy.rs
@@ -5,8 +5,9 @@
 
 use hring::{
     h1, Body, BodyChunk, Encoder, ExpectResponseHeaders, HeadersExt, Responder, Response,
-    ResponseDone, RollMut, ServerDriver,
+    ResponseDone, ServerDriver,
 };
+use hring_buffet::RollMut;
 use http::StatusCode;
 use std::{cell::RefCell, future::Future, net::SocketAddr, rc::Rc};
 use tracing::debug;
diff --git a/test-crates/hring-h2spec/src/main.rs b/test-crates/hring-h2spec/src/main.rs
index 3511a23a..5bf45b15 100644
--- a/test-crates/hring-h2spec/src/main.rs
+++ b/test-crates/hring-h2spec/src/main.rs
@@ -4,10 +4,11 @@
 use std::{collections::VecDeque, net::SocketAddr, path::PathBuf, rc::Rc};
 
 use hring::{
+    buffet::RollMut,
     http::{StatusCode, Version},
     tokio_uring::{self, net::TcpListener},
     Body, BodyChunk, Encoder, ExpectResponseHeaders, Headers, Request, Responder, Response,
-    ResponseDone, RollMut, ServerDriver,
+    ResponseDone, ServerDriver,
 };
 use tokio::process::Command;
 use tracing::{error, info};