libp2p: strangers reads apart

1 files changed, 25 insertions, 6 deletions

diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs
index 2e670bf..3c4ddf5 100644
--- a/src/transport/libp2p/mod.rs
+++ b/src/transport/libp2p/mod.rs
@@ -152,6 +152,9 @@ const STREAM_TIMEOUT: Duration = Duration::from_secs(10);
 const MIN_RATE: u64 = 1 << 20;
 /// Inbound streams read at once.
 const READS: usize = 32;
+/// Of peers not admitted (`transport::Admission`), apart: their streams never take
+/// members' turns. One more is refused, unread (a peer's HELLO is sent again).
+const STRANGER_READS: usize = 8;
 /// Addresses kept per peer, and peers with addresses at most.
 const ADDRESSES_PER_PEER: usize = 8;
 /// Addresses known not to be a peer's (`Book::wrong`), per peer at most.
@@ -643,7 +646,7 @@ impl Transport for Libp2pTransport {
         let _ = self.admission.set(admission);
     }
 
-    /// Reads inbound streams (`READS` at once) and handles their messages one at a
+    /// Reads inbound streams (`READS` at once, `STRANGER_READS` of peers not admitted) and handles their messages one at a
     /// time, in the order they were read; PINGs and PONGs at once (`is_probe`).
     async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
         let Some(mut incoming) = self.incoming.lock().await.take() else {
@@ -652,21 +655,37 @@ impl Transport for Libp2pTransport {
         let handler: Arc<HandlerFn> = Arc::from(handler);
         let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS);
         let reads = Arc::new(Semaphore::new(READS));
+        let stranger_reads = Arc::new(Semaphore::new(STRANGER_READS));
         let probes = Arc::clone(&handler);
         let admission = Arc::clone(&self.admission);
         tokio::spawn(async move {
             while let Some((peer, stream)) = incoming.next().await {
-                let Ok(permit) = Arc::clone(&reads).acquire_owned().await else {
-                    break;
-                };
-                let sender = sender.clone();
-                let probes = Arc::clone(&probes);
                 let limit = crate::transport::Admission::limit(
                     admission.get(),
                     SCHEME,
                     &peer.to_string(),
                     MAX_MESSAGE,
                 );
+                // A stranger's stream, unread while strangers' turns are taken: a
+                // connection that opens streams and sends nothing on them can't hold up
+                // members' messages, nor other strangers' for longer than a stream's
+                // timeout.
+                let permit = if limit < MAX_MESSAGE {
+                    match Arc::clone(&stranger_reads).try_acquire_owned() {
+                        Ok(permit) => permit,
+                        Err(_) => {
+                            debug!(peer = %Addressed(SCHEME, &peer), "Refusing a stream: strangers' reads all taken");
+                            continue;
+                        }
+                    }
+                } else {
+                    let Ok(permit) = Arc::clone(&reads).acquire_owned().await else {
+                        break;
+                    };
+                    permit
+                };
+                let sender = sender.clone();
+                let probes = Arc::clone(&probes);
                 tokio::spawn(async move {
                     match read_message(stream, limit).await {
                         Ok(body) => match wire::decode(&body) {