From c1e5745147537d7a45b5c6979ed9962eecefcd2b Mon Sep 17 00:00:00 2001 From: mp0rta <3p0rta26@gmail.com> Date: Tue, 6 Oct 2026 09:43:08 +0900 Subject: [PATCH 01/33] feat(mq-h3): crate skeleton, h3wire submodule, pass-through decorator --- .gitmodules | 3 + Cargo.lock | 18 ++ Cargo.toml | 2 +- crates/mq-h3/Cargo.toml | 18 ++ crates/mq-h3/src/lib.rs | 295 ++++++++++++++++++++++++++++++ crates/mq-h3/src/queue.rs | 118 ++++++++++++ crates/mq-h3/tests/passthrough.rs | 217 ++++++++++++++++++++++ third_party/h3wire | 1 + 8 files changed, 671 insertions(+), 1 deletion(-) create mode 100644 crates/mq-h3/Cargo.toml create mode 100644 crates/mq-h3/src/lib.rs create mode 100644 crates/mq-h3/src/queue.rs create mode 100644 crates/mq-h3/tests/passthrough.rs create mode 160000 third_party/h3wire diff --git a/.gitmodules b/.gitmodules index b0e229a2..b5cde45b 100644 --- a/.gitmodules +++ b/.gitmodules @@ -2,3 +2,6 @@ path = third_party/xquic url = https://github.com/mp0rta/xquic.git branch = mqvpn-main +[submodule "third_party/h3wire"] + path = third_party/h3wire + url = https://github.com/mp0rta/h3wire.git diff --git a/Cargo.lock b/Cargo.lock index 730fc95a..59f9b213 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -377,6 +377,13 @@ dependencies = [ "tracing", ] +[[package]] +name = "h3wire" +version = "0.1.0" +dependencies = [ + "proptest", +] + [[package]] name = "hashbrown" version = "0.17.1" @@ -526,6 +533,17 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "mq-h3" +version = "0.0.4" +dependencies = [ + "h3wire", + "log", + "mq-runtime", + "mq-transport-api", + "proptest", +] + [[package]] name = "mq-http" version = "0.0.4" diff --git a/Cargo.toml b/Cargo.toml index 2f3433b9..f2a51d1c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = [ "crates/xquic-sys", "crates/mq-linux", "crates/mq-wire", "crates/mq-http", "crates/mq-wire-difftest", "crates/mq-transport-api", "crates/mq-transport", "crates/mq-runtime", - "crates/mq-proxy", "crates/mqproxy", "crates/mq-integration", + "crates/mq-h3", "crates/mq-proxy", "crates/mqproxy", "crates/mq-integration", ] exclude = ["crates/mq-wire/fuzz", "crates/mq-proxy/fuzz", "crates/mq-http/fuzz"] # cargo-fuzz crates are their own workspaces diff --git a/crates/mq-h3/Cargo.toml b/crates/mq-h3/Cargo.toml new file mode 100644 index 00000000..52273871 --- /dev/null +++ b/crates/mq-h3/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "mq-h3" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true + +[dependencies] +h3wire = { path = "../../third_party/h3wire/h3wire" } +mq-transport-api = { path = "../mq-transport-api" } +log = { workspace = true } + +[features] +test-support = [] + +[dev-dependencies] +mq-runtime = { path = "../mq-runtime", features = ["test-support"] } +proptest = { workspace = true } diff --git a/crates/mq-h3/src/lib.rs b/crates/mq-h3/src/lib.rs new file mode 100644 index 00000000..af3f1d1c --- /dev/null +++ b/crates/mq-h3/src/lib.rs @@ -0,0 +1,295 @@ +//! `H3Wire`: a `TransportOps` decorator serving HTTP/3 from h3wire (adoption spec §4.1). +//! +//! C1 scope: the pass-through decorator and the outer event queue. H3 conns served from +//! h3wire (`H3Wire::new`) are filled in by the later tasks. +#![forbid(unsafe_code)] + +mod queue; + +use mq_transport_api::{ + ConnConfig, ConnId, ConnStats, ConnectError, DatagramError, Error, Event, H3Header, H3ReqId, + H3ReqInfo, PathError, PathId, StreamError, StreamId, StreamInfo, Time, Transmit, TransportOps, + TxKey, +}; +use queue::OutQueue; +use std::collections::HashSet; +use std::net::SocketAddr; + +/// Bootstrap read size for a request stream (adoption spec §4.4). +pub const BOOT_READ: usize = 4096; + +pub struct H3Wire { + inner: T, + /// `false`: create no H3 state; every op and event passes through (adoption spec §4.1). + active: bool, + queue: OutQueue, + /// Requests this wrapper holds; always empty until request state lands (later tasks). + reqs: HashSet, +} + +impl H3Wire { + /// Serves H3 conns from h3wire (inner must be `h3_backend: Raw`). + pub fn new(inner: T) -> Self { + Self::make(inner, true) + } + + /// Creates no H3 state: every op and event passes through unfiltered (inner may be xqc_h3). + pub fn passthrough(inner: T) -> Self { + Self::make(inner, false) + } + + fn make(inner: T, active: bool) -> Self { + H3Wire { + inner, + active, + queue: OutQueue::default(), + reqs: HashSet::new(), + } + } + + pub fn inner(&self) -> &T { + &self.inner + } + + pub fn inner_mut(&mut self) -> &mut T { + &mut self.inner + } + + pub fn into_inner(self) -> T { + self.inner + } + + /// Moves the inner queue into ours; runs at the end of every method that takes `now` + /// (adoption spec §4.1). + fn drive_inner(&mut self, _now: Time) { + while let Some(e) = self.inner.poll_event() { + self.queue.push(e); + } + } + + /// An active wrapper holds no request it did not create: anything else is stale. + fn known(&self, r: H3ReqId) -> bool { + self.reqs.contains(&r) + } +} + +/// Runs a forwarding op, then drains the inner queue. +macro_rules! fwd { + ($self:ident, $now:ident, $e:expr) => {{ + let out = $e; + $self.drive_inner($now); + out + }}; +} + +impl TransportOps for H3Wire { + fn recv_datagram(&mut self, now: Time, local: SocketAddr, peer: SocketAddr, data: &[u8]) { + fwd!(self, now, self.inner.recv_datagram(now, local, peer, data)) + } + + fn drive(&mut self, now: Time) { + fwd!(self, now, self.inner.drive(now)) + } + + fn pending_transmit(&self, out: &mut Vec) { + self.inner.pending_transmit(out) + } + + fn peek_transmit(&mut self, key: TxKey) -> Option> { + self.inner.peek_transmit(key) + } + + fn transmit_done(&mut self, key: TxKey, datagrams: usize) { + self.inner.transmit_done(key, datagrams) + } + + fn resume_pending(&self) -> bool { + self.inner.resume_pending() + } + + /// Pops our own queue; stale readiness/creation events are dropped, close events never + /// (adoption spec §4.1). + fn poll_event(&mut self) -> Option { + let Self { + inner, + active, + queue, + reqs, + } = self; + queue.pop(|e| match e { + Event::NewStream(_, s, _) + | Event::StreamReadable(s) + | Event::StreamWritable(s) + | Event::StreamPeerReset(s, _) + | Event::StreamStopSending(s, _) => inner.stream_info(*s).is_ok(), + // The inner transport already filtered passthrough H3 events. + Event::H3Request(_, r) | Event::H3Readable(r) | Event::H3Writable(r) => { + !*active || reqs.contains(r) + } + _ => true, + }) + } + + fn next_timeout(&self) -> Option