refuse unsupported quic client streams without closing the connection

CHANGELOG.md

@@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- **The QUIC broker refuses a client stream it does not support instead of closing the connection** (#207). Any client-opened bidirectional stream that was not a flow discard closed the connection with `ERROR_PROTOCOL_L0` (0xB4), including user-defined flows (0x14) and client data flows, which MQoQ §9.20 allows on bidirectional streams. A user-defined flow on a unidirectional stream was read as MQTT, so depending on its bytes it closed the connection as a malformed packet or was left open. As §9.10 describes, only the stream is now refused, and the connection stays up:
+  - a user-defined flow, or a client data flow on a bidirectional stream, with `ERROR_FLOW_REFUSED` (0xBE);
+  - a server data flow header, or a client data flow whose flow ID has the server bit set, with `ERROR_BAD_FLOW_ID` (0xB6);
+  - a control flow header on any stream but the control stream, or a bidirectional stream without a flow header, with `ERROR_STREAM_TYPE` (0xB5).
+
+  A refused bidirectional stream gets STOP_SENDING and RESET_STREAM with the code, a unidirectional one STOP_SENDING. Unidirectional streams without a flow header are still read as MQTT.
+- **Discarding a flow now removes its subscriptions** (#207). A discard removed the flow's state but left its flow-bound subscriptions, so the broker kept delivering to a flow the client had discarded. §9.21.2 counts subscriptions as part of the flow state that §9.16 says to discard. The broker now removes them before it answers the discard with FIN.
+- **Only a flow header followed directly by FIN is a discard** (§9.16). A bidirectional stream whose header carried the discard flags was treated as a discard even when MQTT data followed, and the data was dropped. Such a stream is now refused with `ERROR_FLOW_REFUSED`.
+
 ## [mqtt5 0.48.0] - 2026-10-11
 
 ### Breaking

crates/mqtt5/src/broker/client_handler/mod.rs

@@ -160,6 +160,7 @@ pub struct ClientHandler {
     pub(super) skip_bridge_forwarding: bool,
     pub(super) pending_target_flow: Option<u64>,
     pub(super) flow_closed_rx: Option<mpsc::Receiver<u64>>,
+    pub(super) flow_discard_rx: Option<mpsc::Receiver<(u64, oneshot::Sender<()>)>>,
     #[cfg(all(not(target_arch = "wasm32"), feature = "transport-quic"))]
     pub(super) flow_registry:
         Option<Arc<tokio::sync::Mutex<crate::session::quic_flow::FlowRegistry>>>,
@@ -281,6 +282,7 @@ impl ClientHandler {
             skip_bridge_forwarding: false,
             pending_target_flow: None,
             flow_closed_rx: None,
+            flow_discard_rx: None,
             #[cfg(all(not(target_arch = "wasm32"), feature = "transport-quic"))]
             flow_registry: None,
             #[cfg(all(not(target_arch = "wasm32"), feature = "transport-quic"))]
@@ -340,6 +342,15 @@ impl ClientHandler {
         self
     }
 
+    #[must_use]
+    pub(crate) fn with_flow_discard_rx(
+        mut self,
+        rx: mpsc::Receiver<(u64, oneshot::Sender<()>)>,
+    ) -> Self {
+        self.flow_discard_rx = Some(rx);
+        self
+    }
+
     #[cfg(all(not(target_arch = "wasm32"), feature = "transport-quic"))]
     #[must_use]
     pub fn with_flow_registry(
@@ -1072,16 +1083,8 @@ impl ClientHandler {
                     }
                 }
 
-                closed_flow_id = async {
-                    if let Some(ref mut rx) = self.flow_closed_rx {
-                        rx.recv().await
-                    } else {
-                        std::future::pending::<Option<u64>>().await
-                    }
-                } => {
-                    if let Some(flow_id) = closed_flow_id {
-                        self.handle_flow_closed(flow_id).await;
-                    }
+                flow_event = Self::next_flow_event(self.flow_closed_rx.as_mut(), self.flow_discard_rx.as_mut()) => {
+                    self.handle_flow_event(flow_event).await;
                 }
 
                 routable = async {

crates/mqtt5/src/broker/client_handler/subscribe.rs

@@ -12,13 +12,19 @@ use crate::packet::Packet;
 use crate::protocol::v5::reason_codes::ReasonCode;
 use crate::validation::{parse_shared_subscription, topic_matches_filter, validate_topic_filter};
 use crate::QoS;
+use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, warn};
 
 use crate::broker::router::{MessageRouter, RoutableMessage, Subscribed, Unsubscribed};
 use crate::broker::session_slot::SessionSlotGuard;
 
 use super::ClientHandler;
 
+pub(super) enum FlowEvent {
+    Closed(u64),
+    Discarded(u64, oneshot::Sender<()>),
+}
+
 struct AcceptedFilter<'a> {
     filter: &'a crate::packet::subscribe::TopicFilter,
     stored: StoredSubscription,
@@ -430,6 +436,47 @@ impl ClientHandler {
         result
     }
 
+    pub(super) async fn next_flow_event(
+        closed_rx: Option<&mut mpsc::Receiver<u64>>,
+        discard_rx: Option<&mut mpsc::Receiver<(u64, oneshot::Sender<()>)>>,
+    ) -> Option<FlowEvent> {
+        let closed = async {
+            match closed_rx {
+                Some(rx) => rx.recv().await.map(FlowEvent::Closed),
+                None => std::future::pending().await,
+            }
+        };
+        let discarded = async {
+            match discard_rx {
+                Some(rx) => rx
+                    .recv()
+                    .await
+                    .map(|(flow_id, done)| FlowEvent::Discarded(flow_id, done)),
+                None => std::future::pending().await,
+            }
+        };
+        tokio::select! {
+            event = closed => event,
+            event = discarded => event,
+        }
+    }
+
+    pub(super) async fn handle_flow_event(&mut self, event: Option<FlowEvent>) {
+        match event {
+            Some(FlowEvent::Closed(flow_id)) => self.handle_flow_closed(flow_id).await,
+            Some(FlowEvent::Discarded(flow_id, done)) => {
+                self.handle_flow_closed(flow_id).await;
+                if done.send(()).is_err() {
+                    debug!(
+                        flow_id,
+                        "discard stream ended before its flow was discarded"
+                    );
+                }
+            }
+            None => {}
+        }
+    }
+
     pub(super) async fn handle_flow_closed(&mut self, flow_id: u64) {
         let Some(ref client_id) = self.client_id else {
             return;