From 8fb0f5d99a83f81df1d5bfc796375a30c6f94abb Mon Sep 17 00:00:00 2001 From: mp0rta <3p0rta26@gmail.com> Date: Wed, 7 Oct 2026 00:03:01 +0900 Subject: [PATCH 1/7] feat(mqproxy): serve HTTP/3 through h3wire --- Cargo.lock | 1 + crates/mqproxy/Cargo.toml | 1 + crates/mqproxy/src/run.rs | 16 +++++++++++----- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fd8a556c..70a5406f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -662,6 +662,7 @@ version = "0.0.4" dependencies = [ "clap", "log", + "mq-h3", "mq-linux", "mq-proxy", "mq-runtime", diff --git a/crates/mqproxy/Cargo.toml b/crates/mqproxy/Cargo.toml index c8863728..05848782 100644 --- a/crates/mqproxy/Cargo.toml +++ b/crates/mqproxy/Cargo.toml @@ -10,6 +10,7 @@ mq-proxy = { path = "../mq-proxy" } mq-runtime = { path = "../mq-runtime" } mq-transport = { path = "../mq-transport" } mq-transport-api = { path = "../mq-transport-api" } +mq-h3 = { path = "../mq-h3" } mq-linux = { path = "../mq-linux" } clap = { workspace = true } log = { workspace = true } diff --git a/crates/mqproxy/src/run.rs b/crates/mqproxy/src/run.rs index efc29334..46c3d8bf 100644 --- a/crates/mqproxy/src/run.rs +++ b/crates/mqproxy/src/run.rs @@ -4,6 +4,7 @@ use crate::cli::{self, Client as ClientArgs, Mode, Resolved, Server as ServerArgs}; use crate::setup_redirect; +use mq_h3::H3Wire; use mq_proxy::server::origin::{build_client_config, native_roots}; use mq_proxy::{client, client::Client, server::Server}; use mq_runtime::driver::{Driver, DriverConfig, StdResolver}; @@ -67,7 +68,12 @@ fn seed() -> u64 { } /// The transport (cert/key loaded here, H3 per `cli::wants_h3`), then `--qlog`. -fn transport(r: &Resolved, role: Role, max_conns: u32, err: String) -> Result { +fn transport( + r: &Resolved, + role: Role, + max_conns: u32, + err: String, +) -> Result, String> { let name = if matches!(role, Role::Client) { "client" } else { @@ -81,7 +87,7 @@ fn transport(r: &Resolved, role: Role, max_conns: u32, err: String) -> Result Result Result { @@ -118,9 +124,9 @@ fn ready(r: &Resolved, line: String) { } /// spec §5.3: `run` → `into_transport` → `Transport::close` with the last `now`. -fn finish(d: Driver, shard: Shard) -> i32 { +fn finish(d: Driver, shard: Shard, A>) -> i32 { let (code, shard) = d.run(shard); - shard.into_transport().close(now()); + shard.into_transport().into_inner().close(now()); code } From 50a07c95968c4c42a2bc57da2b4cfbc132f9865c Mon Sep 17 00:00:00 2001 From: mp0rta <3p0rta26@gmail.com> Date: Wed, 7 Oct 2026 00:24:40 +0900 Subject: [PATCH 2/7] fix(h3): coalesce FIN with final request headers --- crates/mq-h3/src/req.rs | 4 +++ crates/mq-h3/src/send.rs | 19 ++++++++--- crates/mq-h3/tests/send.rs | 66 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 84 insertions(+), 5 deletions(-) diff --git a/crates/mq-h3/src/req.rs b/crates/mq-h3/src/req.rs index 8e03cd61..180a8237 100644 --- a/crates/mq-h3/src/req.rs +++ b/crates/mq-h3/src/req.rs @@ -91,6 +91,9 @@ pub(crate) struct Req { pub(crate) stream_closed: bool, /// The DATA frame `h3_send_body` started and has not written whole (adoption spec §4.5). pub(crate) frame: Option, + /// HEADERS were sent with `fin=true`; the raw FIN rides their final bytes, so + /// suppress h3wire's corresponding empty `FinishStream` action. + pub(crate) coalesced_header_fin: bool, /// `h3_finish` (or a bare `fin`) is waiting for `send_data(0, true)` to stop being /// `Blocked` behind queued HEADERS bytes. pub(crate) finish_latched: bool, @@ -135,6 +138,7 @@ impl Req { fin_read: false, stream_closed: false, frame: None, + coalesced_header_fin: false, finish_latched: false, writable_wanted: false, send_stopped: false, diff --git a/crates/mq-h3/src/send.rs b/crates/mq-h3/src/send.rs index 378150d1..b2c0d952 100644 --- a/crates/mq-h3/src/send.rs +++ b/crates/mq-h3/src/send.rs @@ -137,6 +137,7 @@ impl H3Wire { let r = conn.h3.send_headers(Q(req.quic_id), &fields, fin); req.core |= r.is_ok(); if r.is_ok() { + req.coalesced_header_fin = fin; conn.core.insert(req.quic_id); } self.drive_inner(now); @@ -402,10 +403,17 @@ impl H3Wire { // A `Blocked` FIN is kept and retried; it survives a peer RESET_STREAM // (adoption spec §3, §4.5). Action::FinishStream(q) => { - if let Some(s) = mq(q) - && self.inner.stream_send(now, s, &[], true) == Err(StreamError::Blocked) - { - conn.fins.insert(s); + if let Some(s) = mq(q) { + let headers_fin = self + .reqs + .get_mut(&req_id(s)) + .is_some_and(|req| std::mem::take(&mut req.coalesced_header_fin)); + if !headers_fin + && self.inner.stream_send(now, s, &[], true) + == Err(StreamError::Blocked) + { + conn.fins.insert(s); + } } } // Direction guard (adoption spec §4.5): the transport ops do not check. @@ -456,8 +464,9 @@ impl H3Wire { let mut req = reqs.get_mut(&req_id(s)).filter(|_| q.is_request()); while let Some(bytes) = conn.h3.poll_send(q).filter(|b| !b.is_empty()) { let len = bytes.len(); + let fin = req.as_ref().is_some_and(|req| req.coalesced_header_fin); // Another error is final for the stream: no retry owed. - let n = match inner.stream_send(now, s, bytes, false) { + let n = match inner.stream_send(now, s, bytes, fin) { Ok(n) => n, Err(e) => { short |= e == StreamError::Blocked; diff --git a/crates/mq-h3/tests/send.rs b/crates/mq-h3/tests/send.rs index 1c2f5ed6..8da1c97b 100644 --- a/crates/mq-h3/tests/send.rs +++ b/crates/mq-h3/tests/send.rs @@ -175,6 +175,72 @@ fn client_request_round_trip() { assert_eq!(body, b"hello"); } +#[test] +fn headers_fin_rides_final_header_bytes() { + let mut r = client(); + let (id, s) = open(&mut r); + assert_eq!( + r.w.h3_send_headers(r.now, id, &request("GET"), true), + Ok(()) + ); + let fin_calls: Vec<_> = + r.h.log() + .into_iter() + .filter_map(|c| match c { + Call::StreamSend { + s: sent, + bytes, + fin: true, + } if sent == s => Some(bytes), + _ => None, + }) + .collect(); + assert_eq!(fin_calls.len(), 1); + assert!(!fin_calls[0].is_empty(), "FIN should ride HEADERS bytes"); + assert_eq!(r.fins(s), 1); + r.pump(); + assert_peer_finished(&r, Q(0)); +} + +#[test] +fn headers_fin_survives_partial_write_once() { + let mut r = client(); + let (id, s) = open(&mut r); + r.limit(s, Some(1)); + assert_eq!( + r.w.h3_send_headers(r.now, id, &request("GET"), true), + Ok(()) + ); + assert_eq!(r.h.sent_bytes(s).len(), 1); + assert_eq!(r.fins(s), 0, "partial headers must not commit FIN"); + + r.limit(s, None); + r.h.push_event(Event::StreamWritable(s)); + r.w.drive(r.now); + assert_eq!(r.fins(s), 1); + let fin_calls: Vec<_> = + r.h.log() + .into_iter() + .filter_map(|c| match c { + Call::StreamSend { + s: sent, + bytes, + fin: true, + } if sent == s => Some(bytes), + _ => None, + }) + .collect(); + assert_eq!( + fin_calls.len(), + 2, + "the unfinished header tail retries with FIN" + ); + assert!(fin_calls.iter().all(|bytes| !bytes.is_empty())); + r.pump(); + assert_peer_finished(&r, Q(0)); + assert_eq!(r.fins(s), 1, "the retry commits exactly one FIN"); +} + #[test] fn send_body_partial_payload() { let (mut r, id, s, head) = posting(); From f5e09825c474edbf0b96525f439c23f202830718 Mon Sep 17 00:00:00 2001 From: mp0rta <3p0rta26@gmail.com> Date: Wed, 7 Oct 2026 00:29:09 +0900 Subject: [PATCH 3/7] test(h3): cover blocked header FIN retry --- crates/mq-h3/tests/send.rs | 49 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/crates/mq-h3/tests/send.rs b/crates/mq-h3/tests/send.rs index 8da1c97b..fa9c6c1c 100644 --- a/crates/mq-h3/tests/send.rs +++ b/crates/mq-h3/tests/send.rs @@ -241,6 +241,55 @@ fn headers_fin_survives_partial_write_once() { assert_eq!(r.fins(s), 1, "the retry commits exactly one FIN"); } +#[test] +fn headers_fin_retries_after_blocked_write_once() { + let mut r = client(); + let (id, s) = open(&mut r); + r.limit(s, Some(0)); + assert_eq!( + r.w.h3_send_headers(r.now, id, &request("GET"), true), + Ok(()) + ); + assert!(r.h.sent_bytes(s).is_empty()); + assert_eq!(r.fins(s), 0, "a blocked write cannot commit FIN"); + let blocked: Vec<_> = + r.h.log() + .into_iter() + .filter_map(|c| match c { + Call::StreamSend { + s: sent, + bytes, + fin: true, + } if sent == s => Some(bytes), + _ => None, + }) + .collect(); + assert_eq!(blocked.len(), 1); + assert!(!blocked[0].is_empty(), "the blocked call carries HEADERS"); + + r.limit(s, None); + r.h.push_event(Event::StreamWritable(s)); + r.w.drive(r.now); + assert_eq!(r.fins(s), 1); + let fin_calls: Vec<_> = + r.h.log() + .into_iter() + .filter_map(|c| match c { + Call::StreamSend { + s: sent, + bytes, + fin: true, + } if sent == s => Some(bytes), + _ => None, + }) + .collect(); + assert_eq!(fin_calls.len(), 2, "one blocked attempt, one retry"); + assert!(fin_calls.iter().all(|bytes| !bytes.is_empty())); + r.pump(); + assert_peer_finished(&r, Q(0)); + assert_eq!(r.fins(s), 1, "the retry commits exactly one FIN"); +} + #[test] fn send_body_partial_payload() { let (mut r, id, s, head) = posting(); From d89407c961c541448a8f9d0a088b27375d113812 Mon Sep 17 00:00:00 2001 From: mp0rta <3p0rta26@gmail.com> Date: Wed, 7 Oct 2026 00:37:40 +0900 Subject: [PATCH 4/7] refactor(transport-api): default bodies for the H3 ops --- crates/mq-runtime/tests/app.rs | 49 +-------- crates/mq-transport-api/src/ops.rs | 171 +++++++++++++++++++++++++---- 2 files changed, 151 insertions(+), 69 deletions(-) diff --git a/crates/mq-runtime/tests/app.rs b/crates/mq-runtime/tests/app.rs index fc251083..4a142314 100644 --- a/crates/mq-runtime/tests/app.rs +++ b/crates/mq-runtime/tests/app.rs @@ -5,9 +5,8 @@ use mq_runtime::{ StreamPreread, Target, TcpEnd, TcpId, TimerId, UdpSocketId, }; use mq_transport_api::{ - ConnConfig, ConnId, ConnStats, ConnectError, DatagramError, Error, Event, H3Header, H3ReqId, - H3ReqInfo, PathError, PathId, SlotId, StreamError, StreamId, StreamInfo, Time, Transmit, - TransportOps, TxKey, + ConnConfig, ConnId, ConnStats, ConnectError, DatagramError, Error, Event, PathError, PathId, + SlotId, StreamError, StreamId, StreamInfo, Time, Transmit, TransportOps, TxKey, }; use std::io; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; @@ -85,50 +84,6 @@ impl TransportOps for Null { fn open_uni(&mut self, _: Time, _: ConnId) -> Result { Err(Error::Role) } - fn open_h3_request(&mut self, _: Time, _: ConnId) -> Result { - Err(Error::Role) - } - fn h3_send_headers( - &mut self, - _: Time, - _: H3ReqId, - _: &[H3Header<'_>], - _: bool, - ) -> Result<(), StreamError> { - Err(StreamError::Stale) - } - fn h3_send_body( - &mut self, - _: Time, - _: H3ReqId, - _: &[u8], - _: bool, - ) -> Result { - Err(StreamError::Stale) - } - fn h3_finish(&mut self, _: Time, _: H3ReqId) -> Result<(), StreamError> { - Err(StreamError::Stale) - } - fn h3_recv_headers( - &mut self, - _: Time, - _: H3ReqId, - _: &mut dyn FnMut(&[u8], &[u8]), - ) -> Result { - Err(StreamError::Stale) - } - fn h3_recv_body( - &mut self, - _: Time, - _: H3ReqId, - _: &mut [u8], - ) -> Result<(usize, bool), StreamError> { - Err(StreamError::Stale) - } - fn h3_reset(&mut self, _: Time, _: H3ReqId) {} - fn h3_req_info(&self, _: H3ReqId) -> Result { - Err(Error::Stale) - } } /// Toy app implementing every §5.4 callback. diff --git a/crates/mq-transport-api/src/ops.rs b/crates/mq-transport-api/src/ops.rs index 905fc111..88cb1271 100644 --- a/crates/mq-transport-api/src/ops.rs +++ b/crates/mq-transport-api/src/ops.rs @@ -82,40 +82,167 @@ pub trait TransportOps { // H3 requests (spec §3.1) /// Client only: a server-role transport returns `Error::Role`. - fn open_h3_request(&mut self, now: Time, conn: ConnId) -> Result; + fn open_h3_request(&mut self, _: Time, _: ConnId) -> Result { + Err(Error::Role) + } /// All-or-error; `Blocked` cannot occur with the vendored xquic (spec §3.1). fn h3_send_headers( &mut self, - now: Time, - r: H3ReqId, - hs: &[H3Header<'_>], - fin: bool, - ) -> Result<(), StreamError>; + _: Time, + _: H3ReqId, + _: &[H3Header<'_>], + _: bool, + ) -> Result<(), StreamError> { + Err(StreamError::Conn) + } /// The `stream_send` contract: `Ok(n)` accepted, FIN only when `n == data.len()`. fn h3_send_body( &mut self, - now: Time, - r: H3ReqId, - data: &[u8], - fin: bool, - ) -> Result; + _: Time, + _: H3ReqId, + _: &[u8], + _: bool, + ) -> Result { + Err(StreamError::Conn) + } /// A bare FIN (spec §3.1). - fn h3_finish(&mut self, now: Time, r: H3ReqId) -> Result<(), StreamError>; + fn h3_finish(&mut self, _: Time, _: H3ReqId) -> Result<(), StreamError> { + Err(StreamError::Conn) + } /// Drains one header section; `Ok(fin)`. `Blocked` when none is pending. fn h3_recv_headers( &mut self, - now: Time, - r: H3ReqId, - each: &mut dyn FnMut(&[u8], &[u8]), - ) -> Result; + _: Time, + _: H3ReqId, + _: &mut dyn FnMut(&[u8], &[u8]), + ) -> Result { + Err(StreamError::Conn) + } /// `(bytes, fin)`; `(0, false)` is `Blocked`. fn h3_recv_body( &mut self, - now: Time, - r: H3ReqId, - buf: &mut [u8], - ) -> Result<(usize, bool), StreamError>; + _: Time, + _: H3ReqId, + _: &mut [u8], + ) -> Result<(usize, bool), StreamError> { + Err(StreamError::Conn) + } /// A no-op on a stale id. - fn h3_reset(&mut self, now: Time, r: H3ReqId); - fn h3_req_info(&self, r: H3ReqId) -> Result; + fn h3_reset(&mut self, _: Time, _: H3ReqId) {} + fn h3_req_info(&self, _: H3ReqId) -> Result { + Err(Error::Role) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::SocketAddr; + + struct RawOnly; + + impl TransportOps for RawOnly { + fn recv_datagram(&mut self, _: Time, _: SocketAddr, _: SocketAddr, _: &[u8]) {} + fn drive(&mut self, _: Time) {} + fn pending_transmit(&self, _: &mut Vec) {} + fn peek_transmit(&mut self, _: TxKey) -> Option> { + None + } + fn transmit_done(&mut self, _: TxKey, _: usize) {} + fn resume_pending(&self) -> bool { + false + } + fn poll_event(&mut self) -> Option { + None + } + fn next_timeout(&self) -> Option