H3wire p3 cutover

crates/mq-h3/src/conn.rs

@@ -74,7 +74,7 @@ impl<T: TransportOps> H3Wire<T> {
         self.service(now, c);
     }
 
-    /// An inner event, active mode. Returns it if it passes through; the raw stream events
+    /// An inner event. 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<Event> {
         match e {

crates/mq-h3/src/lib.rs

@@ -1,6 +1,6 @@
 //! `H3Wire<T>`: 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
+//! Runs one h3wire `Connection` per H3 conn and serves the
 //! requests from it.
 #![forbid(unsafe_code)]
 
@@ -26,8 +26,6 @@ pub const BOOT_READ: usize = 4096;
 
 pub struct H3Wire<T> {
     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<H3ReqId, Req>,
@@ -49,20 +47,10 @@ pub struct H3Wire<T> {
 }
 
 impl<T: TransportOps> H3Wire<T> {
-    /// Serves H3 conns from h3wire (inner must be `h3_backend: Raw`).
+    /// Serves H3 conns over the raw transport.
     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(),
@@ -141,11 +129,7 @@ impl<T: TransportOps> H3Wire<T> {
             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)
-            };
+            let out = self.on_event(now, e);
             if let Some(e) = out {
                 self.queue.push(e);
             }
@@ -192,11 +176,7 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
     /// stale between that drain and this pop, so the same checks run again here.
     fn poll_event(&mut self) -> Option<Event> {
         let Self {
-            inner,
-            active,
-            queue,
-            reqs,
-            ..
+            inner, queue, reqs, ..
         } = self;
         queue.pop(|e| match e {
             Event::ConnEstablished(c)
@@ -209,9 +189,6 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
             | 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),
@@ -228,7 +205,7 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
 
     fn connect(&mut self, now: Time, cfg: &ConnConfig) -> Result<ConnId, ConnectError> {
         let r = self.inner.connect(now, cfg);
-        if let (true, ConnProto::H3, Ok(c)) = (self.active, cfg.proto, r) {
+        if let (ConnProto::H3, Ok(c)) = (cfg.proto, r) {
             self.add_conn(now, c, h3wire::Role::Client);
         }
         self.drive_inner(now);
@@ -314,13 +291,10 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
         self.inner.datagram_recv(conn, buf)
     }
 
-    // H3 ops. Active: served from h3wire, never forwarded; an id not held is stale
+    // H3 ops are served from h3wire; an id not held is stale
     // (adoption spec §4.1).
     fn open_h3_request(&mut self, now: Time, conn: ConnId) -> Result<H3ReqId, Error> {
-        if self.active {
-            return self.open_req(now, conn);
-        }
-        fwd!(self, now, self.inner.open_h3_request(now, conn))
+        self.open_req(now, conn)
     }
 
     fn h3_send_headers(
@@ -330,11 +304,8 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
         hs: &[H3Header<'_>],
         fin: bool,
     ) -> Result<(), StreamError> {
-        if self.active {
-            self.touch(r);
-            return self.send_headers(now, r, hs, fin);
-        }
-        fwd!(self, now, self.inner.h3_send_headers(now, r, hs, fin))
+        self.touch(r);
+        self.send_headers(now, r, hs, fin)
     }
 
     fn h3_send_body(
@@ -344,19 +315,13 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
         data: &[u8],
         fin: bool,
     ) -> Result<usize, StreamError> {
-        if self.active {
-            self.touch(r);
-            return self.send_body(now, r, data, fin);
-        }
-        fwd!(self, now, self.inner.h3_send_body(now, r, data, fin))
+        self.touch(r);
+        self.send_body(now, r, data, fin)
     }
 
     fn h3_finish(&mut self, now: Time, r: H3ReqId) -> Result<(), StreamError> {
-        if self.active {
-            self.touch(r);
-            return self.finish(now, r);
-        }
-        fwd!(self, now, self.inner.h3_finish(now, r))
+        self.touch(r);
+        self.finish(now, r)
     }
 
     fn h3_recv_headers(
@@ -365,11 +330,8 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
         r: H3ReqId,
         each: &mut dyn FnMut(&[u8], &[u8]),
     ) -> Result<bool, StreamError> {
-        if self.active {
-            self.touch(r);
-            return self.recv_headers(now, r, each);
-        }
-        fwd!(self, now, self.inner.h3_recv_headers(now, r, each))
+        self.touch(r);
+        self.recv_headers(now, r, each)
     }
 
     fn h3_recv_body(
@@ -382,32 +344,23 @@ impl<T: TransportOps> TransportOps for H3Wire<T> {
         {
             self.max_buf = self.max_buf.max(buf.len());
         }
-        if self.active {
-            self.touch(r);
-            return self.recv_body(now, r, buf);
-        }
-        fwd!(self, now, self.inner.h3_recv_body(now, r, buf))
+        self.touch(r);
+        self.recv_body(now, r, buf)
     }
 
     fn h3_reset(&mut self, now: Time, r: H3ReqId) {
-        if self.active {
-            self.touch(r);
-            return self.reset(now, r);
-        }
-        fwd!(self, now, self.inner.h3_reset(now, r))
+        self.touch(r);
+        self.reset(now, r)
     }
 
     fn h3_req_info(&self, r: H3ReqId) -> Result<H3ReqInfo, Error> {
-        if self.active {
-            // Cached at request start: the inner slot is gone before StreamClosed.
-            return match self.reqs.get(&r) {
-                Some(q) if q.known => Ok(H3ReqInfo {
-                    conn: q.conn,
-                    quic_id: q.quic_id,
-                }),
-                _ => Err(Error::Stale),
-            };
+        // Cached at request start: the inner slot is gone before StreamClosed.
+        match self.reqs.get(&r) {
+            Some(q) if q.known => Ok(H3ReqInfo {
+                conn: q.conn,
+                quic_id: q.quic_id,
+            }),
+            _ => Err(Error::Stale),
         }
-        self.inner.h3_req_info(r)
     }
 }

crates/mq-h3/src/recv.rs

@@ -431,7 +431,6 @@ impl<T: TransportOps> H3Wire<T> {
         if req.known {
             let close = H3Close {
                 stats: req.stats(conn_err),
-                unread: None,
             };
             self.queue.push(Event::H3Closed(id, Box::new(close)));
         }

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<InFlight>,
+    /// 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,

crates/mq-h3/src/send.rs

@@ -95,7 +95,7 @@ impl<T: TransportOps> H3Wire<T> {
     /// Client request start (adoption spec §4.3): the request exists from `open_stream`;
     /// h3wire holds state for it once `send_headers` succeeds.
     pub(crate) fn open_req(&mut self, now: Time, c: ConnId) -> Result<H3ReqId, Error> {
-        // The xqc_h3 backend's order (`reserve_local`): role, conn, protocol.
+        // Request admission order: role, conn, protocol.
         let r = match self.conns.get(&c) {
             _ if self.server => Err(Error::Role),
             None if self.inner.conn_stats(c).is_ok() => Err(Error::Other), // a raw conn
@@ -137,6 +137,7 @@ impl<T: TransportOps> H3Wire<T> {
         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);
@@ -185,7 +186,7 @@ impl<T: TransportOps> H3Wire<T> {
         // A retained request (client only) accepts and discards like after SendStopped,
         // so a running upload pump cannot abort the response (adoption spec §4.3).
         if req.send_stopped || self.conns[&req.conn].gone {
-            // RFC 9114 §4.1: the client's response keeps flowing; xqc_h3 resets the server.
+            // RFC 9114 §4.1: the client's response keeps flowing after cancelling the upload.
             return if client {
                 Ok(data.len())
             } else {
@@ -283,7 +284,7 @@ impl<T: TransportOps> H3Wire<T> {
     }
 
     /// `h3_finish`: latches the finish intent and returns `Ok` while HEADERS bytes are
-    /// queued, as xqc_h3 does (adoption spec §4.5).
+    /// queued (adoption spec §4.5).
     pub(crate) fn finish(&mut self, now: Time, id: H3ReqId) -> Result<(), StreamError> {
         let client = self.client_of(id)?;
         let r = self.body(now, id, &[], true, client).map(drop);
@@ -402,10 +403,17 @@ impl<T: TransportOps> H3Wire<T> {
                 // 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<T: TransportOps> H3Wire<T> {
             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;

crates/mq-h3/tests/common/mod.rs

@@ -345,8 +345,8 @@ impl Rig {
         }
     }
 
-    /// A RESET_STREAM on `q` from a peer stack that resets whatever its h3 state (e.g.
-    /// xqc_h3, even after its FIN); the h3wire peer is not told and its later bytes on `q`
+    /// A RESET_STREAM on `q` from a peer stack that resets even after its FIN;
+    /// the h3wire peer is not told and its later bytes on `q`
     /// are dropped.
     pub fn peer_raw_reset(&mut self, q: u64, code: u64) {
         let Some(s) = self.stream_of(q) else {

crates/mq-h3/tests/lifecycle.rs

@@ -77,14 +77,11 @@ fn started(r: &mut Rig) -> H3ReqId {
     ids[0]
 }
 
-/// The `H3Closed` ids in `ev`, each asserting `unread: None`.
+/// The `H3Closed` ids in `ev`, with their close stats.
 fn closed(ev: &[Event]) -> Vec<H3ReqId> {
     ev.iter()
         .filter_map(|e| match e {
-            Event::H3Closed(id, c) => {
-                assert_eq!(c.unread, None);
-                Some(*id)
-            }
+            Event::H3Closed(id, _) => Some(*id),
             _ => None,
         })
         .collect()