fix: multistream-select negotiation for outbound webrtc substreams

Author: haikoschol

.gitignore

@@ -1 +1,3 @@
 /target
+.idea
+

src/multistream_select/dialer_select.rs

@@ -357,12 +357,25 @@ impl WebRtcDialerState {
         &mut self,
         payload: Vec<u8>,
     ) -> Result<HandshakeResult, crate::error::NegotiationError> {
-        let Message::Protocols(protocols) =
-            Message::decode(payload.into()).map_err(|_| ParseError::InvalidData)?
-        else {
-            return Err(crate::error::NegotiationError::MultistreamSelectError(
-                NegotiationError::Failed,
-            ));
+
+        let protocols = match Message::decode(payload.into()) {
+            Ok(Message::Header(HeaderLine::V1)) => {
+                self.state = HandshakeState::WaitingProtocol;
+                return Ok(HandshakeResult::NotReady);
+            }
+            Ok(Message::Protocol(protocol)) => vec![protocol],
+            Ok(Message::Protocols(protocols)) => protocols,
+            Ok(Message::NotAvailable) => {
+                return match &self.state {
+                    HandshakeState::WaitingProtocol =>
+                        Err(error::NegotiationError::MultistreamSelectError(
+                            NegotiationError::Failed
+                        )),
+                    _ => Err(error::NegotiationError::StateMismatch),
+                }
+            }
+            Ok(Message::ListProtocols) => return Err(error::NegotiationError::StateMismatch),
+            Err(_) => return Err(error::NegotiationError::ParseError(ParseError::InvalidData)),
         };
 
         let mut protocol_iter = protocols.into_iter();
@@ -410,7 +423,9 @@ mod tests {
     use super::*;
     use crate::multistream_select::listener_select_proto;
     use std::time::Duration;
+    use bytes::BufMut;
     use tokio::net::{TcpListener, TcpStream};
+    use crate::multistream_select::protocol::MSG_MULTISTREAM_1_0;
 
     #[tokio::test]
     async fn select_proto_basic() {
@@ -805,17 +820,19 @@ mod tests {
     }
 
     #[test]
-    fn register_response_invalid_message() {
-        // send only header line
+    fn register_response_header_only() {
         let mut bytes = BytesMut::with_capacity(32);
+        bytes.put_u8(MSG_MULTISTREAM_1_0.len() as u8);
+
         let message = Message::Header(HeaderLine::V1);
         message.encode(&mut bytes).map_err(|_| Error::InvalidData).unwrap();
 
         let (mut dialer_state, _message) =
             WebRtcDialerState::propose(ProtocolName::from("/13371338/proto/1"), vec![]).unwrap();
 
         match dialer_state.register_response(bytes.freeze().to_vec()) {
-            Err(error::NegotiationError::MultistreamSelectError(NegotiationError::Failed)) => {}
+            Ok(HandshakeResult::NotReady) => {},
+            Err(err) => panic!("unexpected error: {:?}", err),
             event => panic!("invalid event: {event:?}"),
         }
     }

src/transport/webrtc/connection.rs

@@ -382,21 +382,32 @@ impl WebRtcConnection {
             target: LOG_TARGET,
             peer = ?self.peer,
             ?channel_id,
+            data_len = ?data.len(),
             "handle opening outbound substream",
         );
 
         let rtc_message = WebRtcMessage::decode(&data)
             .map_err(|err| SubstreamError::NegotiationError(err.into()))?;
-        let message = rtc_message.payload.ok_or(SubstreamError::NegotiationError(
+        let payload = rtc_message.payload.ok_or(SubstreamError::NegotiationError(
             ParseError::InvalidData.into(),
         ))?;
 
-        let HandshakeResult::Succeeded(protocol) = dialer_state.register_response(message)? else {
+        // All multistream-select messages are length-prefixed. Since this code path is not using
+        // multistream_select::protocol::MessageIO, we need to decode and remove the length here.
+        let remaining: &[u8] = &payload;
+        let (len, tail) = unsigned_varint::decode::usize(remaining).
+            map_err(|_| SubstreamError::NegotiationError(
+                ParseError::InvalidData.into(),
+            ))?;
+
+        let message = tail[..len].to_vec();
+
+    let HandshakeResult::Succeeded(protocol) = dialer_state.register_response(message)? else {
             tracing::trace!(
                 target: LOG_TARGET,
                 peer = ?self.peer,
                 ?channel_id,
-                "multisteam-select handshake not ready",
+                "multistream-select handshake not ready",
             );
 
             self.channels.insert(
@@ -631,7 +642,7 @@ impl WebRtcConnection {
             protocol: protocol.to_string(),
         });
 
-        self.rtc.channel(channel_id).unwrap().set_buffered_amount_low_threshold(1024);
+        // self.rtc.channel(channel_id).unwrap().set_buffered_amount_low_threshold(1024);
 
         tracing::trace!(
             target: LOG_TARGET,
@@ -803,7 +814,8 @@ impl WebRtcConnection {
                 },
                 event = self.handles.next() => match event {
                     None => unreachable!(),
-                    Some((channel_id, None | Some(SubstreamEvent::Close))) => {
+                    Some((_, None)) => {}
+                    Some((channel_id, Some(SubstreamEvent::Close))) => {
                         tracing::trace!(
                             target: LOG_TARGET,
                             peer = ?self.peer,