diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c1e82524..e5208af1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,7 +43,9 @@ jobs: run: sudo apt-get update && sudo apt-get install -y cmake libevent-dev libcurl4-openssl-dev libnghttp2-dev golang - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -68,7 +70,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | @@ -105,7 +107,9 @@ jobs: run: sudo apt-get update && sudo apt-get install -y cmake libevent-dev libcurl4-openssl-dev libnghttp2-dev golang - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -128,7 +132,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | @@ -201,7 +205,9 @@ jobs: run: sudo apt-get update && sudo apt-get install -y cmake libevent-dev libcurl4-openssl-dev libnghttp2-dev golang - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -224,7 +230,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | @@ -278,7 +284,9 @@ jobs: ca-certificates - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -301,7 +309,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | @@ -352,7 +360,9 @@ jobs: run: sudo apt-get update && sudo apt-get install -y cmake libevent-dev libnghttp2-dev golang pkg-config - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -375,7 +385,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index a5c72de4..b2d06624 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -35,7 +35,9 @@ jobs: # extraction. Same cache keys as ci.yml to ride the existing caches. --- - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -58,7 +60,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | diff --git a/.github/workflows/perf-weekly.yml b/.github/workflows/perf-weekly.yml index d3c980a9..b5f2ee31 100644 --- a/.github/workflows/perf-weekly.yml +++ b/.github/workflows/perf-weekly.yml @@ -45,7 +45,9 @@ jobs: - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -69,7 +71,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' diff --git a/.github/workflows/perf.yml b/.github/workflows/perf.yml index 66801436..1b256398 100644 --- a/.github/workflows/perf.yml +++ b/.github/workflows/perf.yml @@ -45,7 +45,9 @@ jobs: - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache BoringSSL (gcc fPIC) id: bssl-cache @@ -69,7 +71,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v2 + key: xquic-${{ steps.xqsha.outputs.sha }}-linux-gcc-release-eventlog-bssl-0ce57bb-v3 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index d78e4b69..7ba4bf0d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -54,7 +54,9 @@ jobs: - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" # Cache keys are arch-qualified: the matrix runs amd64 + arm64, and a # non-arch key would let arm64 restore an amd64 BoringSSL/xquic build. @@ -79,7 +81,7 @@ jobs: path: | third_party/xquic/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-${{ matrix.arch }}-gcc-release-bssl-0ce57bb-v1 + key: xquic-${{ steps.xqsha.outputs.sha }}-${{ matrix.arch }}-gcc-release-bssl-0ce57bb-v2 - name: Build xquic if: steps.xquic-cache.outputs.cache-hit != 'true' run: | diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 5b25e22f..9655c602 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -135,6 +135,8 @@ jobs: timeout-minutes: 15 steps: - uses: actions/checkout@v6 + with: + submodules: recursive - uses: taiki-e/install-action@v2 with: tool: cargo-deny,cargo-audit @@ -186,7 +188,9 @@ jobs: echo "$HOME/.cargo/bin" >> "$GITHUB_PATH" - name: Compute xquic SHA id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache xquic + BoringSSL (build-xquic.sh) uses: actions/cache@v5 with: @@ -194,7 +198,7 @@ jobs: third_party/xquic/build third_party/xquic/third_party/boringssl/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-build-xquic-sh-v1 + key: xquic-${{ steps.xqsha.outputs.sha }}-build-xquic-sh-v2 - name: Build the C mqproxy (interop peer) run: | scripts/build-xquic.sh @@ -274,7 +278,9 @@ jobs: - name: Compute xquic SHA if: github.event_name != 'push' id: xqsha - run: echo "sha=$(git submodule status third_party/xquic | awk '{print $1}' | tr -d '+-')" >> "$GITHUB_OUTPUT" + run: | + sha=$(git -c safe.directory="$GITHUB_WORKSPACE" rev-parse HEAD:third_party/xquic) + echo "sha=$sha" >> "$GITHUB_OUTPUT" - name: Cache xquic + BoringSSL (build-xquic.sh) if: github.event_name != 'push' uses: actions/cache@v5 @@ -283,7 +289,7 @@ jobs: third_party/xquic/build third_party/xquic/third_party/boringssl/build third_party/xquic/include/xquic/xqc_configure.h - key: xquic-${{ steps.xqsha.outputs.sha }}-build-xquic-sh-v1 + key: xquic-${{ steps.xqsha.outputs.sha }}-build-xquic-sh-v2 - name: Build the C mqproxy (baseline) if: github.event_name != 'push' run: | 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..fd8a556c 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" @@ -535,8 +553,10 @@ name = "mq-integration" version = "0.0.4" dependencies = [ "h2", + "h3wire", "hyper", "log", + "mq-h3", "mq-http", "mq-integration", "mq-linux", 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/conn.rs b/crates/mq-h3/src/conn.rs new file mode 100644 index 00000000..478db536 --- /dev/null +++ b/crates/mq-h3/src/conn.rs @@ -0,0 +1,238 @@ +//! One `h3wire::Connection` per H3 conn, and the raw events it consumes (adoption spec §4.2, +//! §4.4). + +use crate::req::{Req, Terminal, req_id}; +use crate::{BOOT_READ, H3Wire}; +use h3wire::{Config, Connection, H3Code, Role, StreamId as Q}; +use mq_transport_api::{ConnId, ConnProto, Event, StreamError, StreamId, Time, TransportOps}; +use std::collections::{BTreeSet, HashMap, HashSet}; + +/// Equals `mq_transport::H3_FIELD_SECTION_MAX`; mq-h3 does not depend on mq-transport. +const FIELD_SECTION_MAX: usize = 64 * 1024; + +/// adoption spec §4.2. +pub(crate) fn h3_config() -> Config { + let mut c = Config::default(); + c.max_encoded_field_section_size = FIELD_SECTION_MAX; + c.max_field_section_size = Some(FIELD_SECTION_MAX as u64); + c.grease = true; + c.enable_connect_protocol = false; + c.h3_datagram = false; + c +} + +pub(crate) struct H3Conn { + pub(crate) h3: Connection, + pub(crate) client: bool, + /// The mq id of each quic id: local and peer uni streams, request streams. + pub(crate) mq: HashMap, + /// Peer uni streams read to FIN or reset: never fed again, since h3wire would take + /// further bytes for a new stream's type. + pub(crate) uni_done: HashSet, + /// Quic ids of the streams that may hold core bytes: our uni streams (kept; they carry + /// SETTINGS, the stream types and any GOAWAY) and the request streams whose HEADERS + /// were queued, until `poll_send` drains them. All the bytes h3wire owns come from + /// `bind_uni`, `send_headers` and GOAWAY, so `flush` visits these instead of scanning + /// every stream with `sendable()`. Ascending, like `sendable()`. + pub(crate) core: BTreeSet, + /// We closed the transport; h3wire's actions and bytes are dropped from then on. + pub(crate) closing: bool, + /// Request streams whose `FinishStream` write was `Blocked`: retried by `service` + /// (adoption spec §4.5). At most one per stream. + pub(crate) fins: HashSet, + /// The transport conn is gone; the conn stays only for its retained requests + /// (adoption spec §4.3 "Connection close"). + pub(crate) gone: bool, + /// h3wire emitted `Event::Closed`. + pub(crate) h3_closed: bool, + /// The `ConnClosed` code (xquic's `i32`): the stats of a request closed without a + /// snapshot (adoption spec §4.6). + pub(crate) conn_err: Option, +} + +impl H3Conn { + pub(crate) fn is_local(&self, q: Q) -> bool { + q.is_client_initiated() == self.client + } +} + +impl H3Wire { + pub(crate) fn add_conn(&mut self, now: Time, c: ConnId, role: Role) { + let conn = H3Conn { + h3: Connection::new(role, h3_config()), + client: role == Role::Client, + mq: HashMap::new(), + uni_done: HashSet::new(), + core: BTreeSet::new(), + closing: false, + fins: HashSet::new(), + gone: false, + h3_closed: false, + conn_err: None, + }; + self.conns.insert(c, conn); + self.service(now, c); + } + + /// An inner event, active mode. Returns it if it passes through; the raw stream events + /// of H3 conns are consumed here (adoption spec §4.1). + pub(crate) fn on_event(&mut self, now: Time, e: Event) -> Option { + match e { + Event::NewConn(c, proto) => { + self.server = true; + if proto == ConnProto::H3 { + self.add_conn(now, c, Role::Server); + } + } + // The fan-out comes before ConnClosed passes through (adoption spec §4.1). + Event::ConnClosed(c, reason) => self.conn_closed(c, reason.code as i32), + Event::NewStream(c, s, info) => { + let Some(conn) = self.conns.get_mut(&c) else { + return Some(e); + }; + conn.mq.insert(info.quic_id, s); + self.streams.insert(s, (c, Q(info.quic_id))); + // A peer request stream (server): h3wire holds its state from the first + // sight; the gateway learns of it at its HEADERS (adoption spec §4.3). + let q = Q(info.quic_id); + if q.is_request() && !conn.client { + let req = Req::new(now, c, s, info.quic_id, false); + self.reqs.insert(req_id(s), req); + } else if !q.is_uni() && !conn.is_local(q) { + // A server-initiated bidi stream: h3wire closes the conn with + // H3_STREAM_CREATION_ERROR (RFC 9114 §6.1). + log_closed(conn.h3.recv(q, &[], false).map(drop)); + self.service(now, c); + } + return None; + } + Event::StreamReadable(s) + | Event::StreamWritable(s) + | Event::StreamPeerReset(s, _) + | Event::StreamStopSending(s, _) + | Event::StreamCloseStats(s, _) + | Event::StreamClosed(s) => { + let Some(&(c, q)) = self.streams.get(&s) else { + return Some(e); + }; + self.on_stream_event(now, c, s, q, e); + self.service(now, c); + return None; + } + _ => {} + } + Some(e) + } + + /// adoption spec §4.3 "Connection close", steps 1, 3 and 5. A client request whose + /// response is complete but undelivered is retained while h3wire is open; every other + /// request the gateway knows is closed now. + fn conn_closed(&mut self, c: ConnId, code: i32) { + let Some(conn) = self.conns.get_mut(&c) else { + return; + }; + conn.gone = true; + conn.conn_err = Some(code); + conn.closing = true; // h3wire parses on; its actions are dropped + conn.fins.clear(); + let retain = conn.client && !conn.h3_closed; + // ponytail: O(streams + reqs of the shard) per conn close; index them by conn if + // shards grow large. + self.streams.retain(|_, (x, _)| *x != c); + for id in self.reqs_of(c) { + let r = self.reqs.get_mut(&id).expect("listed above"); + r.stream_closed = true; + let complete = r.carry_fin || r.terminal == Some(Terminal::Finished); + if !(retain && r.known && !r.handed && complete) { + self.close_req(id); + } + } + self.settle_conn(c); + } + + /// Drops a gone conn once no retained request is left (adoption spec §4.3 step 5). + pub(crate) fn settle_conn(&mut self, c: ConnId) { + if self.conns.get(&c).is_some_and(|x| x.gone) && !self.reqs.values().any(|r| r.conn == c) { + let mut conn = self.conns.remove(&c).expect("checked"); + conn.h3.transport_closed(); + } + } + + fn on_stream_event(&mut self, now: Time, c: ConnId, s: StreamId, q: Q, e: Event) { + if q.is_request() && self.on_req_event(now, c, s, q, &e) { + return; + } + let conn = self + .conns + .get_mut(&c) + .expect("streams only name live conns"); + let peer_uni = q.is_uni() && !conn.is_local(q); + match e { + Event::StreamReadable(_) if peer_uni => self.read_uni(now, c, s, q), + Event::StreamPeerReset(_, code) => { + if peer_uni && conn.uni_done.insert(s) { + log_closed(conn.h3.stream_reset_received(q, H3Code(code))); + } + // Retirement probe (adoption spec §3). + if peer_uni { + self.read_uni(now, c, s, q); + } + } + Event::StreamStopSending(_, code) => { + log_closed(conn.h3.stop_sending_received(q, H3Code(code))); + } + Event::StreamClosed(_) => { + conn.mq.remove(&q.0); + conn.uni_done.remove(&s); + self.streams.remove(&s); + } + // Writable: `service` flushes. CloseStats of a non-request stream: ignored. + _ => {} + } + } + + /// Reads a peer uni stream eagerly to the transport's end (adoption spec §4.4). + fn read_uni(&mut self, now: Time, c: ConnId, s: StreamId, q: Q) { + let mut buf = [0u8; BOOT_READ]; + loop { + let r = self.inner.stream_recv(now, s, &mut buf); + let conn = self + .conns + .get_mut(&c) + .expect("streams only name live conns"); + match r { + Ok((n, fin)) => { + if !conn.uni_done.contains(&s) { + // A uni stream always consumes everything. + log_closed(conn.h3.recv(q, &buf[..n], fin).map(drop)); + } + if fin { + conn.uni_done.insert(s); + } + if fin || n == 0 { + return; + } + } + Err(StreamError::Reset) => { + // adoption spec §4.4: a deliberate uni-only shortcut. The code only matters + // on a critical stream, which closes the conn whatever it is, so the reset + // is passed now with REQUEST_CANCELLED. Request streams instead follow the + // deferred "reset, code pending" rule (Task C3). + if conn.uni_done.insert(s) { + log_closed(conn.h3.stream_reset_received(q, H3Code::REQUEST_CANCELLED)); + } + return; + } + Err(_) => return, + } + } + } +} + +/// `ConnectionError` is a state notification; the wire effect is a queued action +/// (adoption spec §5.1). +pub(crate) fn log_closed(r: Result<(), h3wire::ConnectionError>) { + if let Err(e) = r { + log::debug!("h3wire: {e}"); + } +} diff --git a/crates/mq-h3/src/lib.rs b/crates/mq-h3/src/lib.rs new file mode 100644 index 00000000..42b6a734 --- /dev/null +++ b/crates/mq-h3/src/lib.rs @@ -0,0 +1,413 @@ +//! `H3Wire`: a `TransportOps` decorator serving HTTP/3 from h3wire (adoption spec §4.1). +//! +//! Active mode (`H3Wire::new`) runs one h3wire `Connection` per H3 conn and serves the +//! requests from it. +#![forbid(unsafe_code)] + +mod conn; +mod queue; +mod recv; +mod req; +mod send; + +use conn::H3Conn; +use mq_transport_api::{ + ConnConfig, ConnId, ConnProto, ConnStats, ConnectError, DatagramError, Error, Event, H3Header, + H3ReqId, H3ReqInfo, PathError, PathId, StreamError, StreamId, StreamInfo, Time, Transmit, + TransportOps, TxKey, +}; +use queue::OutQueue; +use req::Req; +use std::collections::{HashMap, 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, + /// Request streams of H3 conns, by request id (adoption spec §4.3). + reqs: HashMap, + conns: HashMap, + /// H3 conns whose h3wire state changed since their last `service`, or that still wait + /// on the transport (a pending FIN, a short core-bytes write): the only ones swept. + dirty: HashSet, + /// Every stream of an H3 conn: its conn and quic id. + streams: HashMap, + /// A `NewConn` was seen: the inner transport is a server. ponytail: learned, not + /// configured; before its first conn a server reports an unknown conn as `Stale`. + server: bool, + /// The largest `h3_recv_body` buf so far: the carry bound (adoption spec §5.4). + #[cfg(feature = "test-support")] + max_buf: usize, + /// `service` runs so far. + #[cfg(feature = "test-support")] + services: u64, +} + +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: HashMap::new(), + conns: HashMap::new(), + dirty: HashSet::new(), + streams: HashMap::new(), + server: false, + #[cfg(feature = "test-support")] + max_buf: 0, + #[cfg(feature = "test-support")] + services: 0, + } + } + + 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 + } + + /// Request `r`'s h3wire state may change: its conn is swept next (adoption spec §4.5). + fn touch(&mut self, r: H3ReqId) { + if let Some(q) = self.reqs.get(&r) { + self.dirty.insert(q.conn); + } + } + + /// The h3wire abort of request `r` (code, source), if it was aborted. + #[cfg(feature = "test-support")] + pub fn debug_abort(&self, r: H3ReqId) -> Option<(h3wire::H3Code, h3wire::AbortSource)> { + match self.reqs.get(&r)?.terminal? { + req::Terminal::Aborted { code, source } => Some((code, source)), + req::Terminal::Finished => None, + } + } + + /// How many times `service` ran. + #[cfg(feature = "test-support")] + pub fn debug_services(&self) -> u64 { + self.services + } + + /// Streams the core-bytes flush visits per service: our uni streams plus the request + /// streams with HEADERS bytes not yet drained. + #[cfg(feature = "test-support")] + pub fn debug_core_streams(&self) -> usize { + self.conns.values().map(|c| c.core.len()).sum() + } + + /// Every request's carry <= max(BOOT_READ, largest buf passed so far), and every + /// connection's h3wire `debug_buffered_bytes() <= debug_bound()` (adoption spec §5.4). + #[cfg(feature = "test-support")] + pub fn debug_bounds_hold(&self) -> bool { + let carry = BOOT_READ.max(self.max_buf); + self.reqs.values().all(|r| r.carry.len() <= carry) + && self + .conns + .values() + .all(|c| c.h3.debug_buffered_bytes() <= c.h3.debug_bound()) + } + + /// Services the dirty H3 conns, then moves the inner queue into ours, consuming what + /// belongs to H3 conns; runs at the end of every method that takes `now` (adoption spec + /// §4.1). + fn drive_inner(&mut self, now: Time) { + // Core bytes and actions after every call that takes `now` (adoption spec §4.5). + // Running first covers what this call changed; each event below services its own + // conn. A conn nothing touched has nothing to run, so only dirty ones are swept. + let cs: Vec = self.dirty.iter().copied().collect(); + for c in cs { + self.service(now, c); + } + while let Some(e) = self.inner.poll_event() { + let out = if self.active { + self.on_event(now, e) + } else { + Some(e) + }; + if let Some(e) = out { + self.queue.push(e); + } + } + } +} + +/// 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). The inner transport filtered at drain time; an object can go + /// stale between that drain and this pop, so the same checks run again here. + fn poll_event(&mut self) -> Option { + let Self { + inner, + active, + queue, + reqs, + .. + } = self; + queue.pop(|e| match e { + Event::ConnEstablished(c) + | Event::NewConn(c, _) + | Event::MpReady(c) + | Event::DatagramReadable(c) + | Event::PathRemoved(c, _) => inner.conn_live(*c), + Event::NewStream(_, s, _) + | Event::StreamReadable(s) + | Event::StreamWritable(s) + | Event::StreamPeerReset(s, _) + | Event::StreamStopSending(s, _) => inner.stream_info(*s).is_ok(), + Event::H3Request(_, r) | Event::H3Readable(r) | Event::H3Writable(r) if !*active => { + inner.h3_req_info(*r).is_ok() + } + // An H3Readable after the receive end was handed over has nothing to deliver. + Event::H3Readable(r) => reqs.get(r).is_some_and(|q| !q.handed), + Event::H3Request(_, r) | Event::H3Writable(r) => reqs.contains_key(r), + Event::ConnClosed(..) + | Event::StreamClosed(_) + | Event::StreamCloseStats(..) + | Event::H3Closed(..) => true, + }) + } + + fn next_timeout(&self) -> Option