Select endpoint STOP_SENDING code after local FIN

Quiche acceptance of a FIN-bearing HTTP/3 write is tracked separately
from open and aborted send directions.
DriverHooks::STOP_SENDING_CODE_AFTER_LOCAL_FIN selects the code when the
driver detects a dropped inbound-body receiver, without implying peer
delivery or acknowledgement.

Servers use H3_NO_ERROR after submitted responses. Clients, queued or
blocked FINs, open responses, and aborted responses keep
H3_REQUEST_CANCELLED. Focused coverage pins body and trailers FIN, queued
and blocked FIN timing, receiver-drop detection timing, local reset, and
peer STOP transitions.

Author: nox

tokio-quiche/src/http3/driver/client.rs

@@ -218,10 +218,8 @@ impl ClientHooks {
         );
 
         if body_finished {
-            // `send_request()` already sent FIN for bodyless requests, so
-            // mark the send side complete and drop the outbound-frame receiver
-            // since no body will be sent.
-            H3Driver::<ClientHooks>::on_fin_sent(&mut stream_ctx)?;
+            // `send_request()` accepted FIN for this bodyless request.
+            stream_ctx.handle_sent_fin();
         }
 
         if let Some(quarter_stream_id) =
@@ -302,6 +300,9 @@ impl DriverHooks for ClientHooks {
     type Command = ClientH3Command;
     type Event = ClientH3Event;
 
+    const STOP_SENDING_CODE_AFTER_LOCAL_FIN: h3::WireErrorCode =
+        h3::WireErrorCode::RequestCancelled;
+
     fn new(_settings: &Http3Settings) -> Self {
         Self {
             pending_requests: BTreeMap::new(),

tokio-quiche/src/http3/driver/hooks.rs

@@ -57,6 +57,11 @@ pub trait DriverHooks: Sized + Send + 'static {
     /// The concrete type is expected to wrap [`H3Command`].
     type Command: From<H3Command> + Send;
 
+    /// Error code used when the driver detects that the application dropped
+    /// its inbound-body receiver before the peer's send direction ended, after
+    /// quiche accepted a FIN-bearing HTTP/3 write on the local send direction.
+    const STOP_SENDING_CODE_AFTER_LOCAL_FIN: h3::WireErrorCode;
+
     /// Initializes the storage for these hooks.
     fn new(settings: &Http3Settings) -> Self;
 

tokio-quiche/src/http3/driver/mod.rs

@@ -535,7 +535,11 @@ impl<H: DriverHooks> H3Driver<H> {
                 Ok(permit) => permit,
                 Err(TrySendError::Closed(())) => {
                     // The channel has closed before we delivered a fin or reset
-                    // to the application.
+                    // to the application. Choose the code from the send state
+                    // when the driver detects the closure: quiche may have
+                    // accepted local FIN after the application dropped this
+                    // receiver. A FIN still queued or blocked leaves the send
+                    // side open.
                     // Release the channel borrow before updating the context.
                     drop(try_reserve_result);
                     match ctx.recv_state {
@@ -544,8 +548,13 @@ impl<H: DriverHooks> H3Driver<H> {
                             // An associated DATAGRAM flow can outlive its body
                             // channel.
                             {
-                                let err =
-                                    h3::WireErrorCode::RequestCancelled as u64;
+                                let wire_err_code = match ctx.send_state() {
+                                    SendState::Finished =>
+                                        H::STOP_SENDING_CODE_AFTER_LOCAL_FIN,
+                                    SendState::Open | SendState::Aborted =>
+                                        h3::WireErrorCode::RequestCancelled,
+                                };
+                                let err = wire_err_code as u64;
                                 let _ = qconn.stream_shutdown(
                                     stream_id,
                                     quiche::Shutdown::Read,
@@ -866,7 +875,9 @@ impl<H: DriverHooks> H3Driver<H> {
                 let len = body.len();
 
                 if *fin {
-                    // Drop the receiver on FIN to prevent more frames.
+                    // Drop the receiver while FIN is pending to prevent more
+                    // frames. SendState remains Open until quiche accepts the
+                    // entire FIN-bearing write.
                     // We cannot use `mpsc::Receiver::close()` because of a
                     // Tokio inconsistency when reading a closed channel.
                     // See https://github.com/tokio-rs/tokio/issues/7631.
@@ -889,7 +900,7 @@ impl<H: DriverHooks> H3Driver<H> {
                     Err(h3::Error::StreamBlocked)
                 } else {
                     if *fin {
-                        Self::on_fin_sent(ctx)?;
+                        Self::on_fin_accepted(ctx)?;
                     }
 
                     Ok(())
@@ -909,7 +920,7 @@ impl<H: DriverHooks> H3Driver<H> {
                 );
 
                 if res.is_ok() {
-                    Self::on_fin_sent(ctx)?;
+                    Self::on_fin_accepted(ctx)?;
                 }
 
                 return res;
@@ -935,22 +946,8 @@ impl<H: DriverHooks> H3Driver<H> {
         Ok(())
     }
 
-    fn on_fin_sent(ctx: &mut StreamCtx) -> h3::Result<()> {
-        match ctx.send_state {
-            SendState::Open => {},
-            SendState::Closed => {
-                debug_assert!(
-                    false,
-                    "FIN completion requires an open send direction"
-                );
-                return Err(h3::Error::InternalError);
-            },
-        }
-        ctx.cancel_recv_wait();
-        ctx.recv = None;
-        ctx.send_state = SendState::Closed;
-        ctx.audit_stats
-            .set_sent_stream_fin(StreamClosureKind::Explicit);
+    fn on_fin_accepted(ctx: &mut StreamCtx) -> h3::Result<()> {
+        ctx.handle_sent_fin();
         if ctx.fully_closed() {
             // Return a TransportError to trigger stream cleanup
             // instead of h3::Error::Done.
@@ -1269,14 +1266,15 @@ impl<H: DriverHooks> H3Driver<H> {
             match recv.try_recv() {
                 Ok(frame) => ctx.queued_frame = Some(frame),
                 Err(TryRecvError::Disconnected) => {
-                    match ctx.send_state {
+                    match ctx.send_state() {
                         SendState::Open => {
                             if ctx.associated_dgram_flow_id.is_none()
                             // An associated DATAGRAM flow can outlive its body
                             // channel.
                             {
-                                // No FIN was written before channel closure.
-                                // Reset because no more data will follow.
+                                // No FIN-bearing write was accepted before
+                                // channel closure. Reset because no more data
+                                // will follow.
                                 let err =
                                     h3::WireErrorCode::RequestCancelled as u64;
                                 let _ = qconn.stream_shutdown(
@@ -1290,7 +1288,7 @@ impl<H: DriverHooks> H3Driver<H> {
                                 }
                             }
                         },
-                        SendState::Closed => {},
+                        SendState::Finished | SendState::Aborted => {},
                     }
                     break;
                 },

tokio-quiche/src/http3/driver/server.rs

@@ -27,6 +27,7 @@
 use std::ops::Deref;
 use std::sync::Arc;
 
+use quiche::h3;
 use tokio::sync::mpsc;
 
 use super::datagram;
@@ -267,6 +268,9 @@ impl DriverHooks for ServerHooks {
     type Command = ServerH3Command;
     type Event = ServerH3Event;
 
+    const STOP_SENDING_CODE_AFTER_LOCAL_FIN: h3::WireErrorCode =
+        h3::WireErrorCode::NoError;
+
     fn new(settings: &Http3Settings) -> Self {
         Self {
             settings_enforcer: settings.into(),

tokio-quiche/src/http3/driver/streams.rs

@@ -34,6 +34,7 @@ use std::task::Wake;
 use std::task::Waker;
 use std::time::Instant;
 
+use datagram_socket::StreamClosureKind;
 use quiche::h3;
 use tokio::sync::mpsc;
 use tokio_util::sync::PollSender;
@@ -156,13 +157,16 @@ impl RecvState {
     }
 }
 
-/// The transport write direction is closed by FIN, reset, or peer STOP.
+/// Local HTTP/3 write-side lifecycle.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) enum SendState {
-    /// Transport writes have not yet finished or been canceled.
+    /// No FIN-bearing write was accepted and the write side was not aborted.
     Open,
-    /// Transport writes ended through FIN, local reset, or peer STOP_SENDING.
-    Closed,
+    /// Quiche accepted a FIN-bearing HTTP/3 write.
+    Finished,
+    /// The write side ended via reset or peer STOP_SENDING without accepted
+    /// FIN.
+    Aborted,
 }
 
 /// Receives coalesced stream notifications and rearms their wake state.
@@ -250,7 +254,8 @@ pub(crate) struct StreamCtx {
     /// Receive-side lifecycle and the sender for local application delivery.
     pub(super) recv_state: RecvState,
     /// Transport write-side lifecycle.
-    pub(super) send_state: SendState,
+    /// Tracks local FIN acceptance or write-side abortion.
+    send_state: SendState,
     /// The flow ID for proxying datagrams over this stream. If `None`,
     /// the stream has no associated DATAGRAM flow.
     pub(crate) associated_dgram_flow_id: Option<u64>,
@@ -297,12 +302,16 @@ impl StreamCtx {
     /// Cleanup requires closed transport writes and closed app delivery.
     pub(super) fn fully_closed(&self) -> bool {
         match (&self.recv_state, self.send_state) {
-            (RecvState::Closed, SendState::Closed) => true,
+            (RecvState::Closed, SendState::Finished | SendState::Aborted) => true,
             (RecvState::Open(_) | RecvState::FinPending(_), _) |
             (RecvState::Closed, SendState::Open) => false,
         }
     }
 
+    pub(super) fn send_state(&self) -> SendState {
+        self.send_state
+    }
+
     /// Schedules the initial poll to register for upstream channel capacity.
     /// Subsequent readiness uses the same coalescing ready queue.
     pub(crate) fn wait_for_send(&mut self) {
@@ -327,6 +336,7 @@ impl StreamCtx {
         self.waiting_for_recv
     }
 
+    #[cfg(test)]
     pub(super) fn cancel_recv_wait(&mut self) {
         self.waiting_for_recv = false;
     }
@@ -365,22 +375,23 @@ impl StreamCtx {
         Ok((recv_ready, send_ready))
     }
 
-    /// Handle the case when we received a STOP_SENDING frame. Note, that
-    /// we'll only learn about a STOP_SENDING frame from the write path.
-    /// Also note, that quiche will automatically send a RESET_STREAM frame
-    /// in response when it receives a STOP_SENDING (as recommended by the
-    /// RFC)
+    /// Records a peer STOP_SENDING observed from the write path.
+    ///
+    /// Quiche automatically sends RESET_STREAM in response, as recommended by
+    /// the RFC.
     pub(crate) fn handle_recvd_stop_sending(&mut self, wire_err_code: u64) {
         // Update stats even if the stream is already closing.
         self.audit_stats
             .set_recvd_stop_sending_error_code(wire_err_code as i64);
 
         match self.send_state {
-            SendState::Closed => {
-                // STOP may arrive after FIN or reset has closed the write side.
+            SendState::Finished | SendState::Aborted => {
+                // A terminal send state stays terminal. In particular, a later
+                // peer STOP_SENDING does not undo quiche's prior acceptance of
+                // local FIN or imply that the peer received or acknowledged it.
                 return;
             },
-            SendState::Open => self.send_state = SendState::Closed,
+            SendState::Open => self.send_state = SendState::Aborted,
         }
 
         // We received a STOP_SENDING frame. This indicates that the
@@ -423,7 +434,7 @@ impl StreamCtx {
     ) -> H3ConnectionResult<()> {
         match self.send_state {
             SendState::Open => {},
-            SendState::Closed => {
+            SendState::Finished | SendState::Aborted => {
                 debug_assert!(
                     false,
                     "A local reset requires an open send direction"
@@ -433,11 +444,25 @@ impl StreamCtx {
         }
         self.audit_stats
             .set_sent_reset_stream_error_code(wire_err_code as i64);
-        self.send_state = SendState::Closed;
+        self.send_state = SendState::Aborted;
         self.waiting_for_recv = false;
         Ok(())
     }
 
+    pub(crate) fn handle_sent_fin(&mut self) {
+        // This records local submission to quiche, not peer receipt or ACK.
+        debug_assert_eq!(
+            self.send_state,
+            SendState::Open,
+            "accepted FIN requires an open send direction"
+        );
+        self.recv = None;
+        self.waiting_for_recv = false;
+        self.send_state = SendState::Finished;
+        self.audit_stats
+            .set_sent_stream_fin(StreamClosureKind::Explicit);
+    }
+
     pub(crate) fn handle_sent_stop_sending(
         &mut self, wire_err_code: u64,
     ) -> H3ConnectionResult<()> {
@@ -812,7 +837,7 @@ mod tests {
         let (mut ctx, _outbound, _inbound) = StreamCtx::new(4, 1, ready_tx);
         cancel(&mut ctx, 42).unwrap();
         assert_eq!(matches!(ctx.recv_state, RecvState::Closed), closes_receive);
-        assert_eq!(ctx.send_state == SendState::Closed, !closes_receive);
+        assert_eq!(ctx.send_state == SendState::Aborted, !closes_receive);
 
         // Only the second call is allowed to reject the closed direction.
         let result = catch_unwind(AssertUnwindSafe(|| cancel(&mut ctx, 43)));