fix(stream): propagate backpressure

protocols/stream/src/behaviour.rs

@@ -32,7 +32,7 @@ impl Default for Behaviour {
 
 impl Behaviour {
     pub fn new() -> Self {
-        let (dial_sender, dial_receiver) = mpsc::channel(0);
+        let (dial_sender, dial_receiver) = mpsc::channel(32);
 
         Self {
             shared: Arc::new(Mutex::new(Shared::new(dial_sender))),

protocols/stream/src/control.rs

@@ -48,7 +48,17 @@ impl Control {
     ) -> Result<Stream, OpenStreamError> {
         tracing::debug!(%peer, "Requesting new stream");
 
-        let mut new_stream_sender = Shared::lock(&self.shared).sender(peer);
+        let (mut new_stream_sender, dial_info) = {
+            let mut shared = Shared::lock(&self.shared);
+            shared.sender(peer)
+        };
+
+        // If we need to dial, do it outside the lock with backpressure
+        if let Some((peer_to_dial, mut dial_sender)) = dial_info {
+            // Propagate backpressure by awaiting the send instead of try_send
+            dial_sender.send(peer_to_dial).await
+                .map_err(|e| io::Error::new(io::ErrorKind::ConnectionReset, e))?;
+        }
 
         let (sender, receiver) = oneshot::channel();
 

protocols/stream/src/shared.rs

@@ -122,7 +122,7 @@ impl Shared {
         }
     }
 
-    pub(crate) fn sender(&mut self, peer: PeerId) -> mpsc::Sender<NewStream> {
+    pub(crate) fn sender(&mut self, peer: PeerId) -> (mpsc::Sender<NewStream>, Option<(PeerId, mpsc::Sender<PeerId>)>) {
         let maybe_sender = self
             .connections
             .iter()
@@ -134,7 +134,7 @@ impl Shared {
             Some(sender) => {
                 tracing::debug!("Returning sender to existing connection");
 
-                sender.clone()
+                (sender.clone(), None)
             }
             None => {
                 tracing::debug!(%peer, "Not connected to peer, initiating dial");
@@ -144,9 +144,8 @@ impl Shared {
                     .entry(peer)
                     .or_insert_with(|| mpsc::channel(0));
 
-                let _ = self.dial_sender.try_send(peer);
-
-                sender.clone()
+                // Return both the pending channel sender and the dial_sender for async use
+                (sender.clone(), Some((peer, self.dial_sender.clone())))
             }
         }
     }

protocols/stream/tests/lib.rs

@@ -78,3 +78,47 @@ async fn dial_errors_are_propagated() {
     assert_eq!(e.kind(), io::ErrorKind::NotConnected);
     assert_eq!("Dial error: no addresses for peer.", e.to_string());
 }
+
+#[tokio::test]
+async fn backpressure_on_many_concurrent_dials() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::builder()
+                .with_default_directive(LevelFilter::DEBUG.into())
+                .from_env()
+                .unwrap(),
+        )
+        .with_test_writer()
+        .try_init();
+
+    let swarm1 = Swarm::new_ephemeral_tokio(|_| stream::Behaviour::new());
+    let control = swarm1.behaviour().new_control();
+
+    tokio::spawn(swarm1.loop_on_next());
+
+    // Spawn many concurrent dial attempts that will all fail
+    // Before the fix: some would silently drop and hang forever
+    // After the fix: all should fail with proper errors (backpressure propagated)
+    let mut handles = vec![];
+
+    for _ in 0..50 {
+        let mut control_clone = control.clone();
+        let handle = tokio::spawn(async move {
+            let result = control_clone.open_stream(PeerId::random(), PROTOCOL).await;
+            // All should fail, none should hang
+            assert!(result.is_err());
+        });
+        handles.push(handle);
+    }
+
+    // All tasks should complete (not hang indefinitely)
+    for handle in handles {
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            handle,
+        )
+        .await
+        .expect("Task should not hang - backpressure should work")
+        .expect("Task should complete successfully");
+    }
+}