stream: fix overflow in StreamMap::size_hint (#8216)

tokio-stream/src/stream_map.rs

@@ -685,15 +685,15 @@ where
     }
 
     fn size_hint(&self) -> (usize, Option<usize>) {
-        let mut ret = (0, Some(0));
+        let mut ret: (usize, Option<usize>) = (0, Some(0));
 
         for (_, stream) in &self.entries {
             let hint = stream.size_hint();
 
-            ret.0 += hint.0;
+            ret.0 = ret.0.saturating_add(hint.0);
 
             match (ret.1, hint.1) {
-                (Some(a), Some(b)) => ret.1 = Some(a + b),
+                (Some(a), Some(b)) => ret.1 = a.checked_add(b),
                 (Some(_), None) => ret.1 = None,
                 _ => {}
             }

tokio-stream/tests/stream_stream_map.rs

@@ -225,6 +225,30 @@ fn size_hint_without_upper() {
     assert_eq!(size_hint, (3, None));
 }
 
+#[test]
+fn size_hint_overflow() {
+    struct Monster;
+
+    impl Stream for Monster {
+        type Item = ();
+
+        fn poll_next(self: Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> Poll<Option<()>> {
+            panic!()
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            (usize::MAX, Some(usize::MAX))
+        }
+    }
+
+    let mut map = StreamMap::new();
+
+    map.insert("a", Monster);
+    map.insert("b", Monster);
+
+    assert_eq!(map.size_hint(), (usize::MAX, None));
+}
+
 #[test]
 fn new_capacity_zero() {
     let map = StreamMap::<&str, stream::Pending<()>>::new();