Bound stream receive reordering window

src/bin/dosh-client.rs

@@ -8110,6 +8110,14 @@ fn accept_stream_data(
 ) -> (Vec<Vec<u8>>, usize, u64) {
     let stream_id = data.stream_id;
     let expected = stream_next_recv_offset.entry(stream_id).or_insert(0);
+    if !dosh::transport::stream_data_within_receive_window(
+        *expected,
+        data.offset,
+        data.bytes.len(),
+        STREAM_INITIAL_WINDOW,
+    ) {
+        return (Vec::new(), 0, *expected);
+    }
     if data.offset < *expected {
         return (Vec::new(), 0, *expected);
     }

src/bin/dosh-server.rs

@@ -3509,6 +3509,14 @@ async fn flush_stream_pending_data_to_client(
 fn accept_stream_data(client: &mut ClientState, data: StreamData) -> (Vec<Vec<u8>>, usize, u64) {
     let stream_id = data.stream_id;
     let expected = client.stream_next_recv_offset.entry(stream_id).or_insert(0);
+    if !dosh::transport::stream_data_within_receive_window(
+        *expected,
+        data.offset,
+        data.bytes.len(),
+        STREAM_INITIAL_WINDOW,
+    ) {
+        return (Vec::new(), 0, *expected);
+    }
     if data.offset < *expected {
         return (Vec::new(), 0, *expected);
     }

src/transport.rs

@@ -630,6 +630,14 @@ impl StreamMux {
     fn accept_data(&mut self, data: StreamData) -> (Vec<Vec<u8>>, usize, u64) {
         let stream_id = data.stream_id;
         let expected = self.next_recv_offset.entry(stream_id).or_insert(0);
+        if !stream_data_within_receive_window(
+            *expected,
+            data.offset,
+            data.bytes.len(),
+            self.config.initial_window,
+        ) {
+            return (Vec::new(), 0, *expected);
+        }
         if data.offset < *expected {
             return (Vec::new(), 0, *expected);
         }
@@ -769,6 +777,25 @@ pub fn observe_retransmit_rtt(srtt: &mut Option<Duration>, sample: Duration) {
     });
 }
 
+pub fn stream_data_within_receive_window(
+    expected_offset: u64,
+    offset: u64,
+    len: usize,
+    receive_window: usize,
+) -> bool {
+    let Ok(len) = u64::try_from(len) else {
+        return false;
+    };
+    let window = receive_window as u64;
+    if len > window {
+        return false;
+    }
+    let Some(end) = offset.checked_add(len) else {
+        return false;
+    };
+    end <= expected_offset.saturating_add(window)
+}
+
 pub struct DoshTransport {
     socket: Arc<UdpSocket>,
     role: SessionRole,
@@ -1205,6 +1232,73 @@ mod tests {
         assert_eq!(second.received_offset, 10);
     }
 
+    #[test]
+    fn far_future_data_is_not_buffered_beyond_receive_window() {
+        let mut mux = StreamMux::new(TransportConfig {
+            initial_window: 5,
+            ..TransportConfig::default()
+        });
+        mux.accept_open(StreamOpen {
+            stream_id: 3,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+
+        let too_far = mux
+            .handle_data(StreamData {
+                stream_id: 3,
+                offset: 5,
+                bytes: b"!".to_vec(),
+            })
+            .unwrap()
+            .unwrap();
+        assert!(too_far.chunks.is_empty());
+        assert_eq!(too_far.consumed, 0);
+        assert_eq!(too_far.received_offset, 0);
+        assert!(!mux.recv_pending.contains_key(&3));
+
+        let in_window = mux
+            .handle_data(StreamData {
+                stream_id: 3,
+                offset: 4,
+                bytes: b"o".to_vec(),
+            })
+            .unwrap()
+            .unwrap();
+        assert!(in_window.chunks.is_empty());
+        assert_eq!(in_window.received_offset, 0);
+        assert!(
+            mux.recv_pending
+                .get(&3)
+                .is_some_and(|pending| pending.contains_key(&4))
+        );
+
+        let oversized_current = mux
+            .handle_data(StreamData {
+                stream_id: 3,
+                offset: 0,
+                bytes: b"abcdef".to_vec(),
+            })
+            .unwrap()
+            .unwrap();
+        assert!(oversized_current.chunks.is_empty());
+        assert_eq!(oversized_current.received_offset, 0);
+    }
+
+    #[test]
+    fn receive_window_helper_rejects_overflowing_ranges() {
+        assert!(stream_data_within_receive_window(0, 4, 1, 5));
+        assert!(!stream_data_within_receive_window(0, 5, 1, 5));
+        assert!(!stream_data_within_receive_window(0, 0, 6, 5));
+        assert!(!stream_data_within_receive_window(
+            u64::MAX - 1,
+            u64::MAX,
+            1,
+            5
+        ));
+    }
+
     #[test]
     fn window_adjust_acks_sent_data_and_releases_credit() {
         let mut mux = StreamMux::new(TransportConfig {