opara · c84ebe1c · c84ebe1c · c84ebe1c
--- a/crates/tor-proto/src/tunnel/reactor/conflux.rs
+++ b/crates/tor-proto/src/tunnel/reactor/conflux.rs
    }
 }

+/// Flatten a `Result<Result<T, E>, E>` into a `Result<T, E>`.
+///
+/// See the nightly [`Result::flatten`].
+fn flatten<T, E>(x: Result<Result<T, E>, E>) -> Result<T, E> {
--- a/crates/tor-proto/src/tunnel/reactor/conflux.rs
+++ b/crates/tor-proto/src/tunnel/reactor/conflux.rs
            // Note: We don't actually use the returned SinkSendable,
            // and continue writing to the SometimesUboundedSink in the reactor :(
            .map(|res| res.map(|res| res.0))
+            // We only return the first ready action as a Future.
+            // Can't use `next()` since it borrows the stream.
+            .into_future()
--- a/crates/tor-proto/src/tunnel/reactor.rs
+++ b/crates/tor-proto/src/tunnel/reactor.rs
                let msg = unwrap_or_shutdown!(self, ret, "control drop")?;
                Some(CircuitAction::HandleControl(msg))
            },
-            res = circ_actions.next().fuse() => {
-                unwrap_or_shutdown!(self, res, "empty conflux set")???
-            }
+            res = circ_actions.fuse() => res?,