fix(stream): garbage-collect deregistered streams in Shared::accept

Cargo.toml

@@ -101,7 +101,7 @@ libp2p-relay = { version = "0.20.0", path = "protocols/relay" }
 libp2p-rendezvous = { version = "0.16.1", path = "protocols/rendezvous" }
 libp2p-request-response = { version = "0.28.1", path = "protocols/request-response" }
 libp2p-server = { version = "0.12.7", path = "misc/server" }
-libp2p-stream = { version = "0.3.0-alpha", path = "protocols/stream" }
+libp2p-stream = { version = "0.3.0-alpha.1", path = "protocols/stream" }
 libp2p-swarm = { version = "0.47.0", path = "swarm" }
 libp2p-swarm-derive = { version = "=0.35.1", path = "swarm-derive" } # `libp2p-swarm-derive` may not be compatible with different `libp2p-swarm` non-breaking releases. E.g. `libp2p-swarm` might introduce a new enum variant `FromSwarm` (which is `#[non-exhaustive]`) in a non-breaking release. Older versions of `libp2p-swarm-derive` would not forward this enum variant within the `NetworkBehaviour` hierarchy. Thus the version pinning is required.
 libp2p-swarm-test = { version = "0.5.0", path = "swarm-test" }

examples/stream/Cargo.toml

@@ -12,7 +12,7 @@ release = false
 anyhow = "1"
 futures = { workspace = true }
 libp2p = { path = "../../libp2p", features = [ "tokio", "quic"] }
-libp2p-stream = { path = "../../protocols/stream", version = "0.3.0-alpha" }
+libp2p-stream = { path = "../../protocols/stream", version = "0.3.0-alpha.1" }
 rand = "0.8"
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }

protocols/stream/CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.3.0-alpha.1
+
+- Garbage-collect deregistered streams in `Stream::accept`.
+  See [PR 5999](https://github.com/libp2p/rust-libp2p/pull/5999).
+
+
 ## 0.3.0-alpha
 
 - Deprecate `void` crate.

protocols/stream/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libp2p-stream"
-version = "0.3.0-alpha"
+version = "0.3.0-alpha.1"
 edition.workspace = true
 rust-version.workspace = true
 description = "Generic stream protocols for libp2p"

protocols/stream/src/shared.rs

@@ -53,6 +53,9 @@ impl Shared {
         &mut self,
         protocol: StreamProtocol,
     ) -> Result<IncomingStreams, AlreadyRegistered> {
+        self.supported_inbound_protocols
+            .retain(|_, sender| !sender.is_closed());
+
         if self.supported_inbound_protocols.contains_key(&protocol) {
             return Err(AlreadyRegistered);
         }