Allow to skip frames for unknown streams (#59)

Author: fafhrd91

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.11.0] - 2025-07-01
+
+* Allow to skip frames for unknown streams
+
 ## [1.10.0] - 2025-06-29
 
 * Generate unique id for simple client

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-h2"
-version = "1.10.0"
+version = "1.11.0"
 license = "MIT OR Apache-2.0"
 authors = ["Nikolay Kim <[email protected]>"]
 description = "An HTTP/2 client and server"
@@ -24,12 +24,12 @@ features = []
 
 [dependencies]
 ntex-net = "2"
-ntex-io = "2.9"
+ntex-io = "2.13"
 ntex-http = "0.1"
 ntex-bytes = "0.1"
 ntex-codec = "0.6"
-ntex-service = "3.4"
-ntex-util = "2.8"
+ntex-service = "3.5"
+ntex-util = "2.12"
 
 bitflags = "2"
 fxhash = "0.2"

src/client/pool.rs

@@ -175,6 +175,7 @@ impl Client {
                         inner.config.clone(),
                         inner.scheme.clone(),
                         inner.authority.clone(),
+                        inner.skip_unknown_streams,
                         storage,
                     );
                     inner.connections.borrow_mut().push(client);
@@ -273,6 +274,7 @@ struct Inner {
     conn_lifetime: Duration,
     disconnect_timeout: Millis,
     max_streams: u32,
+    skip_unknown_streams: bool,
     scheme: Scheme,
     config: crate::Config,
     authority: ByteString,
@@ -311,6 +313,7 @@ impl ClientBuilder {
             conn_lifetime: Duration::from_secs(0),
             disconnect_timeout: Millis(15_000),
             max_streams: 100,
+            skip_unknown_streams: false,
             minconn: 1,
             maxconn: 16,
             scheme: Scheme::HTTP,
@@ -358,6 +361,14 @@ impl ClientBuilder {
         self
     }
 
+    /// Do not return error for frames for unknown streams.
+    ///
+    /// This includes pending resets, data and window update frames.
+    pub fn skip_unknown_streams(mut self) -> Self {
+        self.0.skip_unknown_streams = true;
+        self
+    }
+
     /// Set max lifetime period for connection.
     ///
     /// Connection lifetime is max lifetime of any opened connection

src/client/simple.rs

@@ -36,6 +36,7 @@ impl SimpleClient {
             config,
             scheme,
             authority,
+            false,
             InflightStorage::default(),
         )
     }
@@ -45,10 +46,11 @@ impl SimpleClient {
         config: Config,
         scheme: Scheme,
         authority: ByteString,
+        skip_unknown_streams: bool,
         storage: InflightStorage,
     ) -> Self {
         let codec = Codec::default();
-        let con = Connection::new(io.get_ref(), codec, config, false);
+        let con = Connection::new(io.get_ref(), codec, config, false, skip_unknown_streams);
         con.set_secure(scheme == Scheme::HTTPS);
 
         let disp = Dispatcher::new(

src/connection.rs

@@ -22,6 +22,7 @@ bitflags::bitflags! {
     #[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
     pub(crate) struct ConnectionFlags: u8 {
         const SETTINGS_PROCESSED      = 0b0000_0001;
+        const UNKNOWN_STREAMS         = 0b0000_0010;
         const DISCONNECT_WHEN_READY   = 0b0000_1000;
         const SECURE                  = 0b0001_0000;
         const STREAM_REFUSED          = 0b0010_0000;
@@ -62,7 +63,13 @@ struct ConnectionState {
 }
 
 impl Connection {
-    pub(crate) fn new(io: IoRef, codec: Codec, config: Config, secure: bool) -> Self {
+    pub(crate) fn new(
+        io: IoRef,
+        codec: Codec,
+        config: Config,
+        secure: bool,
+        skip_streams: bool,
+    ) -> Self {
         // send preface
         if !config.is_server() {
             let _ = io.with_write_buf(|buf| buf.extend_from_slice(&consts::PREFACE));
@@ -94,6 +101,15 @@ impl Connection {
 
         let remote_frame_size = Cell::new(codec.send_frame_size());
 
+        let mut flags = if secure {
+            ConnectionFlags::SECURE
+        } else {
+            ConnectionFlags::empty()
+        };
+        if !skip_streams {
+            flags.insert(ConnectionFlags::UNKNOWN_STREAMS);
+        }
+
         let state = Rc::new(ConnectionState {
             codec,
             remote_frame_size,
@@ -113,11 +129,7 @@ impl Connection {
             local_pending_reset: Default::default(),
             remote_window_sz: Cell::new(frame::DEFAULT_INITIAL_WINDOW_SIZE),
             error: Cell::new(None),
-            flags: Cell::new(if secure {
-                ConnectionFlags::SECURE
-            } else {
-                ConnectionFlags::empty()
-            }),
+            flags: Cell::new(flags),
         });
         let con = Connection(state);
 
@@ -408,7 +420,9 @@ impl Connection {
         }
 
         // Add ids to pending queue
-        self.0.local_pending_reset.add(id, &self.0.local_config);
+        if flags.contains(ConnectionFlags::UNKNOWN_STREAMS) {
+            self.0.local_pending_reset.add(id, &self.0.local_config);
+        }
     }
 
     pub(crate) fn recv_half(&self) -> RecvHalfConnection {
@@ -420,6 +434,12 @@ impl Connection {
     }
 }
 
+impl ConnectionState {
+    fn err_unknown_streams(&self) -> bool {
+        self.flags.get().contains(ConnectionFlags::UNKNOWN_STREAMS)
+    }
+}
+
 impl RecvHalfConnection {
     pub(crate) fn tag(&self) -> &'static str {
         self.0.io.tag()
@@ -455,8 +475,9 @@ impl RecvHalfConnection {
         frm: Headers,
     ) -> Result<Option<(StreamRef, Message)>, Either<ConnectionError, StreamErrorInner>> {
         let id = frm.stream_id();
+        let is_server = self.0.local_config.is_server();
 
-        if self.0.local_config.is_server() && !id.is_client_initiated() {
+        if is_server && !id.is_client_initiated() {
             return Err(Either::Left(ConnectionError::InvalidStreamId(
                 "Invalid id in received headers frame",
             )));
@@ -467,7 +488,9 @@ impl RecvHalfConnection {
                 Ok(item) => Ok(item.map(move |msg| (stream, msg))),
                 Err(kind) => Err(Either::Right(StreamErrorInner::new(stream, kind))),
             }
-        } else if !self.0.local_config.is_server() && self.0.local_pending_reset.is_pending(id) {
+        } else if !is_server
+            && (!self.0.err_unknown_streams() || self.0.local_pending_reset.is_pending(id))
+        {
             // if client and no stream, then it was closed
             self.encode(frame::Reset::new(id, frame::Reason::STREAM_CLOSED));
             Ok(None)
@@ -543,7 +566,9 @@ impl RecvHalfConnection {
                 Ok(item) => Ok(item.map(move |msg| (stream, msg))),
                 Err(kind) => Err(Either::Right(StreamErrorInner::new(stream, kind))),
             }
-        } else if self.0.local_pending_reset.is_pending(frm.stream_id()) {
+        } else if !self.0.err_unknown_streams()
+            || self.0.local_pending_reset.is_pending(frm.stream_id())
+        {
             self.encode(frame::Reset::new(
                 frm.stream_id(),
                 frame::Reason::STREAM_CLOSED,
@@ -684,11 +709,17 @@ impl RecvHalfConnection {
                 .map_err(|kind| Either::Right(StreamErrorInner::new(stream, kind)))
         } else if self.0.local_pending_reset.is_pending(frm.stream_id()) {
             Ok(())
-        } else {
+        } else if self.0.err_unknown_streams() {
             log::trace!("Unknown WINDOW_UPDATE {:?}", frm);
             Err(Either::Left(ConnectionError::UnknownStream(
                 "WINDOW_UPDATE",
             )))
+        } else {
+            self.encode(frame::Reset::new(
+                frm.stream_id(),
+                frame::Reason::STREAM_CLOSED,
+            ));
+            Ok(())
         }
     }
 
@@ -724,9 +755,11 @@ impl RecvHalfConnection {
             )))
         } else if self.0.local_pending_reset.remove(id) {
             self.update_rst_count()
-        } else {
+        } else if self.0.err_unknown_streams() {
             self.update_rst_count()?;
             Err(Either::Left(ConnectionError::UnknownStream("RST_STREAM")))
+        } else {
+            Ok(())
         }
     }
 

src/server/service.rs

@@ -136,7 +136,13 @@ where
 
         // create h2 codec
         let codec = Codec::default();
-        let con = Connection::new(io.get_ref(), codec.clone(), inner.config.clone(), true);
+        let con = Connection::new(
+            io.get_ref(),
+            codec.clone(),
+            inner.config.clone(),
+            true,
+            false,
+        );
         let con2 = con.clone();
 
         // start protocol dispatcher
@@ -249,7 +255,7 @@ where
 
     // create h2 codec
     let codec = Codec::default();
-    let con = Connection::new(io.get_ref(), codec.clone(), config.clone(), true);
+    let con = Connection::new(io.get_ref(), codec.clone(), config.clone(), true, false);
     let con2 = con.clone();
 
     // start protocol dispatcher