Refactor WebRTC code (#51)

This PR refactors the WebRTC code into something that actually works.
The `WebRtcConnection` code is split into `OpeningWebRtcConnection` and
`WebRtcConnection`. The former only deals with the Noise handshake
whereas the latter implements a typical connection event loop, dealing
with inbound/outbound substreams and data exchange. The PR also
implements a new substream type for WebRTC which is implemented using
`tokio::sync::mpsc::{Sender, Receiver}` since `str0m` doesn't expose a
substream type. The `multistream-select` code used by WebRTC is
refactored and its test coverage is increased.

This PR also updates `str0m` to the latest version but during testing it
was discovered that there are issues with `ChannelId`/SCTP stream id
reuse and those have to be reported to upstream and fixed. They're are
not blockers for merging this PR though because the fixes should be
contained to `str0m` and for litep2p it would only be a version bump if
the fixes are accepted.

The WebRTC implementation (with pending fixes to `str0m`) has been
tested with smoldot and the code works. IPFS Ping, Kademlia and Identify
all work, opening and negotiating notification protocols work and block
requests work. There are, however, two issues that I was able to find:

1) The first "notification" smoldot reads from the substream is not
actually a notification but the substream handshake which results in
"protocol-error":
```
[network] protocol-error; peer_id=12D3KooWNGCW63xu9tGh1WxWEazHs6UgMdTjrLSos89NrVSCh9bz, error=BadBlockAnnounce(DecodeBlockAnnounceError(Verify))
[network] protocol-error; peer_id=12D3KooWNGCW63xu9tGh1WxWEazHs6UgMdTjrLSos89NrVSCh9bz, error=BadGrandpaNotification(DecodeGrandpaNotificationError(Verify))
```

It looks like smoldot is still able to receive block announcements and
GRANDPA notifications so I'm not entirely what the issue here is.

2) There is another issues with justifications but I don't see how it
could be related to WebRTC
```
[sync-service-westend2] finality-proof-verify-error; error=FinalityVerify(UnknownTargetBlock { block_number: 19772928, block_hash: [22, 163, 175, 94, 114, 75, 142, 174, 83, 146, 11, 251, 184, 119, 212, 76, 34, 85, 144, 47, 83, 251, 83, 86, 152, 141, 204, 104, 253, 236, 21, 48] }), sender=12D3KooWNGCW63xu9tGh1WxWEazHs6UgMdTjrLSos89NrVSCh9bz
[sync-service-westend2] Error while verifying justification: Justification targets a block (#19772928) that isn't in the chain.
```

This happens on the first try, gossip is closed, then it's opened again
and it's able to start warp sync.

---------

Co-authored-by: Dmitry Markin <[email protected]>
Co-authored-by: Alexandru Vasile <[email protected]>

Author: 3 people

Cargo.lock

@@ -36,7 +36,7 @@ simple-dns = "0.5.3"
 smallvec = "1.10.0"
 snow = { version = "0.9.3", features = ["ring-resolver"], default-features = false }
 socket2 = { version = "0.5.5", features = ["all"] }
-str0m = "0.2.0"
+str0m = "0.4.1"
 thiserror = "1.0.39"
 tokio-stream = "0.1.12"
 tokio-tungstenite = { version = "0.20.0", features = ["rustls-tls-native-roots"] }

Cargo.toml

@@ -24,7 +24,9 @@ use crate::{
     codec::unsigned_varint::UnsignedVarint,
     error::{self, Error},
     multistream_select::{
-        protocol::{HeaderLine, Message, MessageIO, Protocol, ProtocolError},
+        protocol::{
+            encode_multistream_message, HeaderLine, Message, MessageIO, Protocol, ProtocolError,
+        },
         Negotiated, NegotiationError, Version,
     },
     types::protocol::ProtocolName,
@@ -224,7 +226,7 @@ where
 }
 
 /// `multistream-select` handshake result for dialer.
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq)]
 pub enum HandshakeResult {
     /// Handshake is not complete, data missing.
     NotReady,
@@ -259,7 +261,6 @@ pub struct DialerState {
     state: HandshakeState,
 }
 
-// TODO: tests
 impl DialerState {
     /// Propose protocol to remote peer.
     ///
@@ -269,29 +270,22 @@ impl DialerState {
         protocol: ProtocolName,
         fallback_names: Vec<ProtocolName>,
     ) -> crate::Result<(Self, Vec<u8>)> {
-        // encode `/multistream-select/1.0.0` header
-        let mut bytes = BytesMut::with_capacity(64);
-        let message = Message::Header(HeaderLine::V1);
-        let _ = message.encode(&mut bytes).map_err(|_| Error::InvalidData)?;
-        let mut header = UnsignedVarint::encode(bytes)?;
-
-        // encode proposed protocol
-        let mut proto_bytes = BytesMut::with_capacity(512);
-        let message = Message::Protocol(Protocol::try_from(protocol.as_bytes()).unwrap());
-        let _ = message.encode(&mut proto_bytes).map_err(|_| Error::InvalidData)?;
-        let proto_bytes = UnsignedVarint::encode(proto_bytes)?;
-
-        // TODO: add fallback names
-
-        header.append(&mut proto_bytes.into());
+        let message = encode_multistream_message(
+            std::iter::once(protocol.clone())
+                .chain(fallback_names.clone())
+                .filter_map(|protocol| Protocol::try_from(protocol.as_ref()).ok())
+                .map(|protocol| Message::Protocol(protocol)),
+        )?
+        .freeze()
+        .to_vec();
 
         Ok((
             Self {
                 protocol,
                 fallback_names,
                 state: HandshakeState::WaitingResponse,
             },
-            header,
+            message,
         ))
     }
 
@@ -328,10 +322,9 @@ impl DialerState {
                         return Ok(HandshakeResult::Succeeded(self.protocol.clone()));
                     }
 
-                    // TODO: zzz
                     for fallback in &self.fallback_names {
                         if fallback.as_bytes() == protocol.as_ref() {
-                            return Ok(HandshakeResult::Succeeded(self.protocol.clone()));
+                            return Ok(HandshakeResult::Succeeded(fallback.clone()));
                         }
                     }
 
@@ -346,3 +339,148 @@ impl DialerState {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn propose() {
+        let (mut dialer_state, message) =
+            DialerState::propose(ProtocolName::from("/13371338/proto/1"), vec![]).unwrap();
+        let message = bytes::BytesMut::from(&message[..]).freeze();
+
+        let Message::Protocols(protocols) = Message::decode(message).unwrap() else {
+            panic!("invalid message type");
+        };
+
+        assert_eq!(protocols.len(), 2);
+        assert_eq!(
+            protocols[0],
+            Protocol::try_from(&b"/multistream/1.0.0"[..])
+                .expect("valid multitstream-select header")
+        );
+        assert_eq!(
+            protocols[1],
+            Protocol::try_from(&b"/13371338/proto/1"[..])
+                .expect("valid multitstream-select header")
+        );
+    }
+
+    #[test]
+    fn propose_with_fallback() {
+        let (mut dialer_state, message) = DialerState::propose(
+            ProtocolName::from("/13371338/proto/1"),
+            vec![ProtocolName::from("/sup/proto/1")],
+        )
+        .unwrap();
+        let message = bytes::BytesMut::from(&message[..]).freeze();
+
+        let Message::Protocols(protocols) = Message::decode(message).unwrap() else {
+            panic!("invalid message type");
+        };
+
+        assert_eq!(protocols.len(), 3);
+        assert_eq!(
+            protocols[0],
+            Protocol::try_from(&b"/multistream/1.0.0"[..])
+                .expect("valid multitstream-select header")
+        );
+        assert_eq!(
+            protocols[1],
+            Protocol::try_from(&b"/13371338/proto/1"[..])
+                .expect("valid multitstream-select header")
+        );
+        assert_eq!(
+            protocols[2],
+            Protocol::try_from(&b"/sup/proto/1"[..]).expect("valid multitstream-select header")
+        );
+    }
+
+    #[test]
+    fn register_response_invalid_message() {
+        // send only header line
+        let mut bytes = BytesMut::with_capacity(32);
+        let message = Message::Header(HeaderLine::V1);
+        let _ = message.encode(&mut bytes).map_err(|_| Error::InvalidData).unwrap();
+
+        let (mut dialer_state, _message) =
+            DialerState::propose(ProtocolName::from("/13371338/proto/1"), vec![]).unwrap();
+
+        match dialer_state.register_response(bytes.freeze().to_vec()) {
+            Err(Error::NegotiationError(error::NegotiationError::MultistreamSelectError(
+                NegotiationError::Failed,
+            ))) => {}
+            event => panic!("invalid event: {event:?}"),
+        }
+    }
+
+    #[test]
+    fn header_line_missing() {
+        // header line missing
+        let mut bytes = BytesMut::with_capacity(256);
+        let message = Message::Protocols(vec![
+            Protocol::try_from(&b"/13371338/proto/1"[..]).unwrap(),
+            Protocol::try_from(&b"/sup/proto/1"[..]).unwrap(),
+        ]);
+        let _ = message.encode(&mut bytes).map_err(|_| Error::InvalidData).unwrap();
+
+        let (mut dialer_state, _message) =
+            DialerState::propose(ProtocolName::from("/13371338/proto/1"), vec![]).unwrap();
+
+        match dialer_state.register_response(bytes.freeze().to_vec()) {
+            Err(Error::NegotiationError(error::NegotiationError::MultistreamSelectError(
+                NegotiationError::Failed,
+            ))) => {}
+            event => panic!("invalid event: {event:?}"),
+        }
+    }
+
+    #[test]
+    fn negotiate_main_protocol() {
+        let message = encode_multistream_message(
+            vec![Message::Protocol(
+                Protocol::try_from(&b"/13371338/proto/1"[..]).unwrap(),
+            )]
+            .into_iter(),
+        )
+        .unwrap()
+        .freeze();
+
+        let (mut dialer_state, _message) = DialerState::propose(
+            ProtocolName::from("/13371338/proto/1"),
+            vec![ProtocolName::from("/sup/proto/1")],
+        )
+        .unwrap();
+
+        match dialer_state.register_response(message.to_vec()) {
+            Ok(HandshakeResult::Succeeded(negotiated)) =>
+                assert_eq!(negotiated, ProtocolName::from("/13371338/proto/1")),
+            _ => panic!("invalid event"),
+        }
+    }
+
+    #[test]
+    fn negotiate_fallback_protocol() {
+        let message = encode_multistream_message(
+            vec![Message::Protocol(
+                Protocol::try_from(&b"/sup/proto/1"[..]).unwrap(),
+            )]
+            .into_iter(),
+        )
+        .unwrap()
+        .freeze();
+
+        let (mut dialer_state, _message) = DialerState::propose(
+            ProtocolName::from("/13371338/proto/1"),
+            vec![ProtocolName::from("/sup/proto/1")],
+        )
+        .unwrap();
+
+        match dialer_state.register_response(message.to_vec()) {
+            Ok(HandshakeResult::Succeeded(negotiated)) =>
+                assert_eq!(negotiated, ProtocolName::from("/sup/proto/1")),
+            _ => panic!("invalid event"),
+        }
+    }
+}