Merge branch 'master' into web-transport-quic

.github/workflows/ci.yml

@@ -310,7 +310,7 @@ jobs:
       - uses: actions/checkout@v4
       - run: wget -q -O- https://github.com/obi1kenobi/cargo-semver-checks/releases/download/v0.31.0/cargo-semver-checks-x86_64-unknown-linux-gnu.tar.gz | tar -xz -C ~/.cargo/bin
         shell: bash
-      - uses: obi1kenobi/cargo-semver-checks-action@c7306483f698c511eaf7416d1bf2e1958c90140f # v2
+      - uses: obi1kenobi/cargo-semver-checks-action@ca26a44cfb670b2078c8f757d06e696a7c3820cf # v2
 
   rustfmt:
     runs-on: ubuntu-latest

Cargo.toml

@@ -82,7 +82,7 @@ libp2p-core = { version = "0.41.3", path = "core" }
 libp2p-dcutr = { version = "0.11.1", path = "protocols/dcutr" }
 libp2p-dns = { version = "0.41.1", path = "transports/dns" }
 libp2p-floodsub = { version = "0.44.0", path = "protocols/floodsub" }
-libp2p-gossipsub = { version = "0.46.2", path = "protocols/gossipsub" }
+libp2p-gossipsub = { version = "0.47.0", path = "protocols/gossipsub" }
 libp2p-identify = { version = "0.45.0", path = "protocols/identify" }
 libp2p-identity = { version = "0.2.9" }
 libp2p-kad = { version = "0.46.0", path = "protocols/kad" }
@@ -92,7 +92,7 @@ libp2p-metrics = { version = "0.14.2", path = "misc/metrics" }
 libp2p-mplex = { version = "0.41.0", path = "muxers/mplex" }
 libp2p-muxer-test-harness = { path = "muxers/test-harness" }
 libp2p-noise = { version = "0.44.0", path = "transports/noise" }
-libp2p-perf = { version = "0.3.1", path = "protocols/perf" }
+libp2p-perf = { version = "0.4.0", path = "protocols/perf" }
 libp2p-ping = { version = "0.44.2", path = "protocols/ping" }
 libp2p-plaintext = { version = "0.41.0", path = "transports/plaintext" }
 libp2p-pnet = { version = "0.24.0", path = "transports/pnet" }
@@ -112,7 +112,7 @@ libp2p-upnp = { version = "0.2.2", path = "protocols/upnp" }
 libp2p-webrtc = { version = "0.7.1-alpha", path = "transports/webrtc" }
 libp2p-webrtc-utils = { version = "0.2.1", path = "misc/webrtc-utils" }
 libp2p-webrtc-websys = { version = "0.3.0-alpha", path = "transports/webrtc-websys" }
-libp2p-websocket = { version = "0.43.1", path = "transports/websocket" }
+libp2p-websocket = { version = "0.43.2", path = "transports/websocket" }
 libp2p-websocket-websys = { version = "0.3.2", path = "transports/websocket-websys" }
 libp2p-webtransport-websys = { version = "0.3.0", path = "transports/webtransport-websys" }
 libp2p-yamux = { version = "0.45.2", path = "muxers/yamux" }

examples/rendezvous/src/bin/rzv-identify.rs

@@ -76,8 +76,12 @@ async fn main() {
             }
             // once `/identify` did its job, we know our external address and can register
             SwarmEvent::Behaviour(MyBehaviourEvent::Identify(identify::Event::Received {
+                info,
                 ..
             })) => {
+                // Register our external address. Needs to be done explicitly
+                // for this case, as it's a local address.
+                swarm.add_external_address(info.observed_addr);
                 if let Err(error) = swarm.behaviour_mut().rendezvous.register(
                     rendezvous::Namespace::from_static("rendezvous"),
                     rendezvous_point,

protocols/gossipsub/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.47.0
+- Add ConnectionError to FromSwarm::ConnectionClosed.
+  See [PR 5485](https://github.com/libp2p/rust-libp2p/pull/5485).
+
 ## 0.46.2
 - Use `web-time` instead of `instant`.
   See [PR 5347](https://github.com/libp2p/rust-libp2p/pull/5347).

protocols/gossipsub/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-gossipsub"
 edition = "2021"
 rust-version = { workspace = true }
 description = "Gossipsub protocol for libp2p"
-version = "0.46.2"
+version = "0.47.0"
 authors = ["Age Manning <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

protocols/gossipsub/src/behaviour/tests.rs

@@ -268,6 +268,7 @@ where
                 connection_id,
                 endpoint: &fake_endpoint,
                 remaining_established: active_connections,
+                cause: None,
             }));
         }
     }

protocols/perf/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.4.0
+- Add ConnectionError to FromSwarm::ConnectionClosed.
+  See [PR 5485](https://github.com/libp2p/rust-libp2p/pull/5485).
+
 ## 0.3.1
 - Use `web-time` instead of `instant`.
   See [PR 5347](https://github.com/libp2p/rust-libp2p/pull/5347).

protocols/perf/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-perf"
 edition = "2021"
 rust-version = { workspace = true }
 description = "libp2p perf protocol implementation"
-version = "0.3.1"
+version = "0.4.0"
 authors = ["Max Inden <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

protocols/perf/src/client/behaviour.rs

@@ -116,6 +116,7 @@ impl NetworkBehaviour for Behaviour {
                 connection_id: _,
                 endpoint: _,
                 remaining_established,
+                ..
             }) => {
                 if remaining_established == 0 {
                     assert!(self.connected.remove(&peer_id));

swarm/CHANGELOG.md

@@ -10,6 +10,8 @@
   See [PR 5347](https://github.com/libp2p/rust-libp2p/pull/5347).
 - Add `#[track_caller]` on all `spawn` wrappers.
   See [PR 5465](https://github.com/libp2p/rust-libp2p/pull/5465).
+- Add ConnectionError to FromSwarm::ConnectionClosed.
+  See [PR 5485](https://github.com/libp2p/rust-libp2p/pull/5485).
 
 ## 0.44.2
 

swarm/src/behaviour.rs

@@ -32,8 +32,8 @@ use crate::connection::ConnectionId;
 use crate::dial_opts::DialOpts;
 use crate::listen_opts::ListenOpts;
 use crate::{
-    ConnectionDenied, ConnectionHandler, DialError, ListenError, THandler, THandlerInEvent,
-    THandlerOutEvent,
+    ConnectionDenied, ConnectionError, ConnectionHandler, DialError, ListenError, THandler,
+    THandlerInEvent, THandlerOutEvent,
 };
 use libp2p_core::{transport::ListenerId, ConnectedPoint, Endpoint, Multiaddr};
 use libp2p_identity::PeerId;
@@ -481,6 +481,7 @@ pub struct ConnectionClosed<'a> {
     pub peer_id: PeerId,
     pub connection_id: ConnectionId,
     pub endpoint: &'a ConnectedPoint,
+    pub cause: Option<&'a ConnectionError>,
     pub remaining_established: usize,
 }
 

swarm/src/lib.rs

@@ -909,6 +909,7 @@ where
                         peer_id,
                         connection_id: id,
                         endpoint: &endpoint,
+                        cause: error.as_ref(),
                         remaining_established: num_established as usize,
                     }));
                 self.pending_swarm_events

swarm/src/test.rs

@@ -301,6 +301,7 @@ where
             connection_id,
             endpoint,
             remaining_established,
+            cause,
         }: ConnectionClosed,
     ) {
         let mut other_closed_connections = self
@@ -350,6 +351,7 @@ where
                 connection_id,
                 endpoint,
                 remaining_established,
+                cause,
             }));
     }
 }

transports/websocket/CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.43.2
+
+- fix: Avoid websocket panic on polling after errors. See [PR 5482].
+
+[PR 5482]: https://github.com/libp2p/rust-libp2p/pull/5482
+
 ## 0.43.1
 
 ## 0.43.0

transports/websocket/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-websocket"
 edition = "2021"
 rust-version = { workspace = true }
 description = "WebSocket transport for libp2p"
-version = "0.43.1"
+version = "0.43.2"
 authors = ["Parity Technologies <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"
@@ -21,6 +21,7 @@ pin-project-lite = "0.2.14"
 rw-stream-sink = { workspace = true }
 soketto = "0.8.0"
 tracing = { workspace = true }
+thiserror = "1.0.61"
 url = "2.5"
 webpki-roots = "0.25"
 

transports/websocket/src/framed.rs

@@ -571,7 +571,7 @@ fn location_to_multiaddr<T>(location: &str) -> Result<Multiaddr, Error<T>> {
 /// The websocket connection.
 pub struct Connection<T> {
     receiver: BoxStream<'static, Result<Incoming, connection::Error>>,
-    sender: Pin<Box<dyn Sink<OutgoingData, Error = connection::Error> + Send>>,
+    sender: Pin<Box<dyn Sink<OutgoingData, Error = quicksink::Error<connection::Error>> + Send>>,
     _marker: std::marker::PhantomData<T>,
 }
 

transports/websocket/src/quicksink.rs

@@ -30,14 +30,6 @@
 //     Ok::<_, io::Error>(stdout)
 // });
 // ```
-//
-// # Panics
-//
-// - If any of the [`Sink`] methods produce an error, the sink transitions
-// to a failure state and none of its methods must be called afterwards or
-// else a panic will occur.
-// - If [`Sink::poll_close`] has been called, no other sink method must be
-// called afterwards or else a panic will be caused.
 
 use futures::{ready, sink::Sink};
 use pin_project_lite::pin_project;
@@ -102,6 +94,15 @@ enum State {
     Failed,
 }
 
+/// Errors the `Sink` may return.
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum Error<E> {
+    #[error("Error while sending over the sink, {0}")]
+    Send(E),
+    #[error("The Sink has closed")]
+    Closed,
+}
+
 pin_project! {
     /// `SinkImpl` implements the `Sink` trait.
     #[derive(Debug)]
@@ -119,7 +120,7 @@ where
     F: FnMut(S, Action<A>) -> T,
     T: Future<Output = Result<S, E>>,
 {
-    type Error = E;
+    type Error = Error<E>;
 
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
         let mut this = self.project();
@@ -135,28 +136,27 @@ where
                     Err(e) => {
                         this.future.set(None);
                         *this.state = State::Failed;
-                        Poll::Ready(Err(e))
+                        Poll::Ready(Err(Error::Send(e)))
                     }
                 }
             }
             State::Closing => match ready!(this.future.as_mut().as_pin_mut().unwrap().poll(cx)) {
                 Ok(_) => {
                     this.future.set(None);
                     *this.state = State::Closed;
-                    panic!("SinkImpl::poll_ready called on a closing sink.")
+                    Poll::Ready(Err(Error::Closed))
                 }
                 Err(e) => {
                     this.future.set(None);
                     *this.state = State::Failed;
-                    Poll::Ready(Err(e))
+                    Poll::Ready(Err(Error::Send(e)))
                 }
             },
             State::Empty => {
                 assert!(this.param.is_some());
                 Poll::Ready(Ok(()))
             }
-            State::Closed => panic!("SinkImpl::poll_ready called on a closed sink."),
-            State::Failed => panic!("SinkImpl::poll_ready called after error."),
+            State::Closed | State::Failed => Poll::Ready(Err(Error::Closed)),
         }
     }
 
@@ -193,7 +193,7 @@ where
                     Err(e) => {
                         this.future.set(None);
                         *this.state = State::Failed;
-                        return Poll::Ready(Err(e));
+                        return Poll::Ready(Err(Error::Send(e)));
                     }
                 },
                 State::Flushing => {
@@ -207,7 +207,7 @@ where
                         Err(e) => {
                             this.future.set(None);
                             *this.state = State::Failed;
-                            return Poll::Ready(Err(e));
+                            return Poll::Ready(Err(Error::Send(e)));
                         }
                     }
                 }
@@ -221,11 +221,10 @@ where
                     Err(e) => {
                         this.future.set(None);
                         *this.state = State::Failed;
-                        return Poll::Ready(Err(e));
+                        return Poll::Ready(Err(Error::Send(e)));
                     }
                 },
-                State::Closed => return Poll::Ready(Ok(())),
-                State::Failed => panic!("SinkImpl::poll_flush called after error."),
+                State::Closed | State::Failed => return Poll::Ready(Err(Error::Closed)),
             }
         }
     }
@@ -253,7 +252,7 @@ where
                     Err(e) => {
                         this.future.set(None);
                         *this.state = State::Failed;
-                        return Poll::Ready(Err(e));
+                        return Poll::Ready(Err(Error::Send(e)));
                     }
                 },
                 State::Flushing => {
@@ -266,7 +265,7 @@ where
                         Err(e) => {
                             this.future.set(None);
                             *this.state = State::Failed;
-                            return Poll::Ready(Err(e));
+                            return Poll::Ready(Err(Error::Send(e)));
                         }
                     }
                 }
@@ -280,11 +279,11 @@ where
                     Err(e) => {
                         this.future.set(None);
                         *this.state = State::Failed;
-                        return Poll::Ready(Err(e));
+                        return Poll::Ready(Err(Error::Send(e)));
                     }
                 },
                 State::Closed => return Poll::Ready(Ok(())),
-                State::Failed => panic!("SinkImpl::poll_closed called after error."),
+                State::Failed => return Poll::Ready(Err(Error::Closed)),
             }
         }
     }
@@ -347,4 +346,31 @@ mod tests {
             assert_eq!(&expected[..], &actual[..])
         });
     }
+
+    #[test]
+    fn error_does_not_panic() {
+        task::block_on(async {
+            let sink = make_sink(io::stdout(), |mut _stdout, _action| async move {
+                Err(io::Error::new(io::ErrorKind::Other, "oh no"))
+            });
+
+            futures::pin_mut!(sink);
+
+            let result = sink.send("hello").await;
+            match result {
+                Err(crate::quicksink::Error::Send(e)) => {
+                    assert_eq!(e.kind(), io::ErrorKind::Other);
+                    assert_eq!(e.to_string(), "oh no")
+                }
+                _ => panic!("unexpected result: {:?}", result),
+            };
+
+            // Call send again, expect not to panic.
+            let result = sink.send("hello").await;
+            match result {
+                Err(crate::quicksink::Error::Closed) => {}
+                _ => panic!("unexpected result: {:?}", result),
+            };
+        })
+    }
 }