fix(swarm): avoid re-extracting completed outbound substreams

swarm/CHANGELOG.md

@@ -6,6 +6,11 @@
 - Raise MSRV to 1.88.0.
   See [PR 6273](https://github.com/libp2p/rust-libp2p/pull/6273).
 
+- Fix `panic!("cannot extract twice")` in `Connection::poll` by selecting
+  a `Waiting` `SubstreamRequested` explicitly instead of the first entry,
+  which could be a stale `Done` left behind by a prior extraction.
+  See [PR 6427](https://github.com/libp2p/rust-libp2p/pull/6427).
+
 ## 0.47.1
 
 - Replace `lru::LruCache` with `hashlink::LruCache`.

swarm/src/connection.rs

@@ -411,7 +411,15 @@ where
                 }
             }
 
-            if let Some(requested_substream) = requested_substreams.iter_mut().next() {
+            // Skip `Done` entries: `extract()` leaves them behind until the
+            // entry's `Future::poll` removes them on the next `poll_next`.
+            // If `extract()` runs before that future has stored its waker,
+            // the `Done` entry can persist and a plain `iter_mut().next()`
+            // would re-extract it and panic with "cannot extract twice".
+            if let Some(requested_substream) = requested_substreams
+                .iter_mut()
+                .find(|r| matches!(r, SubstreamRequested::Waiting { .. }))
+            {
                 match muxing.poll_outbound_unpin(cx)? {
                     Poll::Pending => {}
                     Poll::Ready(substream) => {
@@ -855,6 +863,36 @@ mod tests {
         ))
     }
 
+    // Regression test for `panic!("cannot extract twice")`: pushing two
+    // entries and calling `extract()` without ever polling them leaves
+    // a `Done` entry that `iter_mut().next()` would re-extract. The
+    // `find(Waiting)` filter must skip past it.
+    #[test]
+    fn iter_mut_skips_done_substream_requested_entries() {
+        let mut requested: FuturesUnordered<SubstreamRequested<(), DeniedUpgrade>> =
+            FuturesUnordered::new();
+        for _ in 0..2 {
+            requested.push(SubstreamRequested::new(
+                (),
+                Duration::from_secs(60),
+                DeniedUpgrade,
+            ));
+        }
+
+        for _ in 0..2 {
+            let _ = requested
+                .iter_mut()
+                .find(|r| matches!(r, SubstreamRequested::Waiting { .. }))
+                .unwrap()
+                .extract();
+        }
+        assert!(
+            requested
+                .iter_mut()
+                .all(|r| matches!(r, SubstreamRequested::Done))
+        );
+    }
+
     #[test]
     fn propagates_changes_to_supported_inbound_protocols() {
         let mut connection = Connection::new(