net: Bound concurrent broadcasts

bin/app/src/plugin/darkirc.rs

@@ -505,7 +505,9 @@ impl DarkIrc {
             error!(target: "darkirc", "Failed inserting new event to DAG: {}", e);
         }
 
-        self.p2p.broadcast(&EventPut(event, vec![])).await;
+        if let Err(e) = self.p2p.broadcast(&EventPut(event, vec![])).await {
+            error!(target: "darkirc", "Event broadcast was not admitted: {e}");
+        }
     }
 
     async fn apply_settings(self_: Arc<Self>, _: BatchGuardPtr) {

bin/darkfid/src/registry/mod.rs

@@ -201,7 +201,7 @@ impl DarkfiMinersRegistryState {
             "Broadcasting new block to network",
         );
         let message = ProposalMessage(proposal);
-        p2p_handler.p2p.broadcast(&message).await;
+        p2p_handler.p2p.broadcast(&message).await?;
 
         Ok(())
     }

bin/darkfid/src/rpc/tx.rs

@@ -154,7 +154,10 @@ impl DarkfiNode {
             return server_error(RpcError::TxSimulationFail, id, None)
         };
 
-        self.p2p_handler.p2p.broadcast(&tx).await;
+        if let Err(e) = self.p2p_handler.p2p.broadcast(&tx).await {
+            error!(target: "darkfid::rpc::tx_broadcast", "Transaction broadcast was not admitted: {e}");
+            return JsonError::new(InternalError, None, id).into()
+        }
         if !self.p2p_handler.p2p.is_connected() {
             warn!(target: "darkfid::rpc::tx_broadcast", "No connected channels to broadcast tx");
         }
@@ -240,7 +243,10 @@ impl DarkfiNode {
                     return JsonError::new(InternalError, None, id).into()
                 }
             };
-            self.p2p_handler.p2p.broadcast(&tx).await;
+            if let Err(e) = self.p2p_handler.p2p.broadcast(&tx).await {
+                error!(target: "darkfid::rpc::tx_rebroadcast_pending", "Transaction broadcast was not admitted: {e}");
+                return JsonError::new(InternalError, None, id).into()
+            }
         }
 
         JsonResponse::new(JsonValue::Boolean(true), id).into()

bin/darkfid/src/task/unknown_proposal.rs

@@ -224,7 +224,17 @@ async fn handle_unknown_proposal(node: &DarkfiNodePtr, channel: u32, proposal: &
 
         // Broadcast proposal to rest nodes
         let message = ProposalMessage(proposal.clone());
-        node.p2p_handler.p2p.broadcast_with_exclude(&message, &[channel.address().clone()]).await;
+        if let Err(e) = node
+            .p2p_handler
+            .p2p
+            .broadcast_with_exclude(&message, &[channel.address().clone()])
+            .await
+        {
+            debug!(
+                target: "darkfid::task::handle_unknown_proposal",
+                "Proposal broadcast was not admitted: {e}"
+            );
+        }
 
         // Notify proposals subscriber
         let enc_prop = JsonValue::String(base64::encode(&serialize_async(proposal).await));
@@ -480,7 +490,9 @@ async fn handle_reorg(
 
     // Broadcast proposal to the network
     let message = ProposalMessage(proposal.clone());
-    node.p2p_handler.p2p.broadcast(&message).await;
+    if let Err(e) = node.p2p_handler.p2p.broadcast(&message).await {
+        debug!(target: "darkfid::task::handle_reorg", "Proposal broadcast was not admitted: {e}");
+    }
 
     // Notify proposals subscriber
     let enc_prop = JsonValue::String(base64::encode(&serialize_async(proposal).await));

bin/darkfid/src/tests/harness.rs

@@ -176,7 +176,7 @@ impl Harness {
             let proposal = Proposal::new(block.clone());
             self.alice.validator.write().await.append_proposal(&proposal, None).await?;
             let message = ProposalMessage(proposal);
-            self.alice.p2p_handler.p2p.broadcast(&message).await;
+            self.alice.p2p_handler.p2p.broadcast(&message).await?;
         }
 
         // Sleep a bit so blocks can be propagated and then

bin/darkirc/src/irc/client.rs

@@ -320,7 +320,11 @@ impl Client {
                                     return Err(e)
                                 }
 
-                                self.server.darkirc.p2p.broadcast(&EventPut(event, blob)).await;
+                                if let Err(e) =
+                                    self.server.darkirc.p2p.broadcast(&EventPut(event, blob)).await
+                                {
+                                    error!("[IRC CLIENT] Event broadcast was not admitted: {e}");
+                                }
                             }
                         }
 

bin/tau/taud/src/main.rs

@@ -355,7 +355,9 @@ async fn start_sync_loop(
                     } else {
                         // Otherwise, broadcast it. Taud runs EventGraph with RLN disabled,
                         // so the blob is intentionally empty.
-                        p2p.broadcast(&EventPut(event, vec![])).await;
+                        if let Err(e) = p2p.broadcast(&EventPut(event, vec![])).await {
+                            error!(target: "taud", "Event broadcast was not admitted: {e}");
+                        }
                     }
                 }
             }

example/dchat/dchatd/src/rpc.rs

@@ -20,7 +20,7 @@ use async_trait::async_trait;
 use darkfi::{net::P2pPtr, system::StoppableTaskPtr};
 use smol::lock::MutexGuard;
 use std::collections::HashSet;
-use tracing::debug;
+use tracing::{debug, error};
 
 use darkfi::rpc::{
     jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
@@ -62,7 +62,9 @@ impl Dchat {
     async fn send(&self, id: i64, params: JsonValue) -> JsonResult {
         let msg = params[0].get::<String>().unwrap().to_string();
         let dchatmsg = DchatMsg { msg };
-        self.p2p.broadcast(&dchatmsg).await;
+        if let Err(e) = self.p2p.broadcast(&dchatmsg).await {
+            error!(target: "dchatd::rpc", "Message broadcast was not admitted: {e}");
+        }
         JsonResponse::new(JsonValue::Boolean(true), id).into()
     }
 

script/evgrd/bin/evgrd.rs

@@ -251,12 +251,12 @@ async fn send_event(stream: &mut Box<dyn PtStream>, daemon: &Daemon) -> Result<(
     }
 
     if !peers_with_matched_version.is_empty() {
-        p2p.broadcast_to(&EventPut(event.clone()), &peers_with_matched_version).await;
+        p2p.broadcast_to(&EventPut(event.clone()), &peers_with_matched_version).await?;
     }
     if !peers_with_different_version.is_empty() {
         let mut event = event;
         event.timestamp /= 1000;
-        p2p.broadcast_to(&EventPut(event), &peers_with_different_version).await;
+        p2p.broadcast_to(&EventPut(event), &peers_with_different_version).await?;
     }
 
     Ok(())

script/research/dam/damd/src/flooder.rs

@@ -209,7 +209,7 @@ async fn flood_bar(p2p: P2pPtr, subscriber: JsonSubscriber, limit: u32) -> Resul
         let notification = format!("Broadcasting bar message: {message}");
         info!(target: "damd::flooder::flood_bar", "{notification}");
         subscriber.notify(vec![JsonValue::String(notification)].into()).await;
-        p2p.broadcast(&Bar { message }).await;
+        p2p.broadcast(&Bar { message }).await?;
         message_index += 1;
 
         // Check limit

script/research/generic-node/src/main.rs

@@ -311,11 +311,11 @@ async fn broadcast_messages(node_id: u64, p2p: P2pPtr) -> Result<()> {
         // Broadcast a generic string message
         let string_msg =
             GenericStringMessage { msg: format!("Hello from node {node_id}({counter})!") };
-        p2p.broadcast(&string_msg).await;
+        p2p.broadcast(&string_msg).await?;
 
         // Broadcast a generic number message
         let number_msg = GenericNumberMessage { num: node_id + counter };
-        p2p.broadcast(&number_msg).await;
+        p2p.broadcast(&number_msg).await?;
 
         // Perform a direct request to each peer and grab their response
         let peers = p2p.hosts().channels();

script/research/net_hashmap.rs

@@ -84,7 +84,7 @@ where
     /// Additionally, this change will be broadcasted to the P2P network.
     pub async fn insert(&mut self, k: K, v: V) -> Result<Option<V>> {
         let message = NetHashMapInsert { k: k.clone(), v: v.clone() };
-        self.p2p.broadcast(&message).await;
+        self.p2p.broadcast(&message).await?;
         Ok(self.hashmap.insert(k, v))
     }
 
@@ -101,7 +101,7 @@ where
         Q: Hash + Eq + Send + Sync + Encodable + Decodable + 'static,
     {
         let message = NetHashMapRemove { k: k.clone() };
-        self.p2p.broadcast(&message).await;
+        self.p2p.broadcast(&message).await?;
         Ok(self.hashmap.remove(&k))
     }
 

src/error.rs

@@ -141,6 +141,9 @@ pub enum Error {
     #[error("Network service stopped")]
     NetworkServiceStopped,
 
+    #[error("P2P broadcast concurrency limit reached")]
+    BroadcastLimitReached,
+
     #[error("Create listener bound to {0} failed")]
     BindFailed(String),
 

src/event_graph/mod.rs

@@ -2757,8 +2757,7 @@ impl EventGraph {
     }
 
     pub async fn static_broadcast(&self, ev: Event, blob: Vec<u8>) -> Result<()> {
-        self.p2p.broadcast(&StaticPut(ev, blob)).await;
-        Ok(())
+        self.p2p.broadcast(&StaticPut(ev, blob)).await
     }
 
     fn static_persist_serialized(&self, ev_id: &blake3::Hash, ev_bytes: &[u8]) -> Result<()> {

src/event_graph/proto.rs

@@ -37,7 +37,7 @@ use darkfi_serial::{
     SerialEncodable,
 };
 use smol::Executor;
-use tracing::{error, warn};
+use tracing::{debug, error, warn};
 
 use super::{
     event::Header,
@@ -1113,10 +1113,14 @@ impl ProtocolEventGraph {
                     (RATELIMIT_SAMPLE_IDX - RATELIMIT_MIN_COUNT)) as u64;
                 msleep(ms).await;
             }
-            self.event_graph
+            if let Err(e) = self
+                .event_graph
                 .p2p
                 .broadcast_with_exclude(&ep, &[self.channel.address().clone()])
-                .await;
+                .await
+            {
+                debug!(target: "event_graph::protocol", "Broadcast was not admitted: {e}");
+            }
         }
     }
 }

src/event_graph/tests.rs

@@ -1422,7 +1422,7 @@ async fn propagation_with_real_blob(ex: Arc<Executor<'static>>) {
     nodes[0].header_dag_insert(vec![event.header.clone()], &dag_name).await.unwrap();
     nodes[0].dag_insert(slice::from_ref(&event), &dag_name).await.unwrap();
     nodes[0].dag_blob_store(&event.id(), &blob).unwrap();
-    nodes[0].p2p.broadcast(&EventPut(event.clone(), blob.clone())).await;
+    nodes[0].p2p.broadcast(&EventPut(event.clone(), blob.clone())).await.unwrap();
 
     sleep(5).await;
 
@@ -1472,7 +1472,7 @@ async fn event_put_ingests_body_after_header_only_sync(ex: Arc<Executor<'static>
     }
     assert!(nodes[4].dag_blob_fetch(&event.id()).unwrap().is_none());
 
-    nodes[0].p2p.broadcast(&EventPut(event.clone(), blob.clone())).await;
+    nodes[0].p2p.broadcast(&EventPut(event.clone(), blob.clone())).await.unwrap();
     sleep(5).await;
 
     let store = nodes[4].dag_store.read().await;
@@ -1500,7 +1500,7 @@ async fn empty_blob_rejected(ex: Arc<Executor<'static>>) {
     let dag_ts = nodes[0].current_genesis.read().await.header.timestamp;
     let event = Event::new(b"unauthenticated".to_vec(), &nodes[0]).await.unwrap();
 
-    nodes[0].p2p.broadcast(&EventPut(event.clone(), vec![])).await;
+    nodes[0].p2p.broadcast(&EventPut(event.clone(), vec![])).await.unwrap();
     sleep(5).await;
 
     for (i, eg) in nodes.iter().enumerate().skip(1) {
@@ -1539,7 +1539,7 @@ async fn malformed_event_rejected_before_rln(ex: Arc<Executor<'static>>) {
     malformed.content.extend_from_slice(b"-tampered");
     assert!(!malformed.content_matches_header());
 
-    nodes[0].p2p.broadcast(&EventPut(malformed.clone(), blob)).await;
+    nodes[0].p2p.broadcast(&EventPut(malformed.clone(), blob)).await.unwrap();
     sleep(5).await;
 
     let epoch = epoch_of(malformed.header.timestamp);
@@ -1584,7 +1584,7 @@ async fn genesis_with_blob_rejected(ex: Arc<Executor<'static>>) {
     let event = Event { header, content: b"forged-genesis".to_vec() };
     let fake_blob = b"this-should-not-be-here".to_vec();
 
-    nodes[0].p2p.broadcast(&EventPut(event.clone(), fake_blob)).await;
+    nodes[0].p2p.broadcast(&EventPut(event.clone(), fake_blob)).await.unwrap();
     sleep(5).await;
 
     for (i, eg) in nodes.iter().enumerate().skip(1) {
@@ -1876,7 +1876,7 @@ async fn fetch_parents_rejects_child_when_parent_body_rejected(ex: Arc<Executor<
         seed_rotating_event_unchecked(eg, &parent, &bad_parent_blob, &dag_name).await;
     }
 
-    nodes[0].p2p.broadcast(&EventPut(child.clone(), child_blob)).await;
+    nodes[0].p2p.broadcast(&EventPut(child.clone(), child_blob)).await.unwrap();
     sleep(5).await;
 
     let store = nodes[4].dag_store.read().await;
@@ -1936,7 +1936,7 @@ async fn dormant_user_can_post_after_long_silence(ex: Arc<Executor<'static>>) {
     nodes[0].header_dag_insert(vec![event.header.clone()], &dag_name).await.unwrap();
     nodes[0].dag_insert(slice::from_ref(&event), &dag_name).await.unwrap();
     nodes[0].dag_blob_store(&event.id(), &blob).unwrap();
-    nodes[0].p2p.broadcast(&EventPut(event.clone(), blob)).await;
+    nodes[0].p2p.broadcast(&EventPut(event.clone(), blob)).await.unwrap();
 
     sleep(5).await;
 

src/net/p2p.rs

@@ -17,9 +17,10 @@
  */
 
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
+    future::Future,
     sync::{
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool, AtomicUsize, Ordering},
         Arc, Weak,
     },
 };
@@ -45,9 +46,9 @@ use super::{
     settings::Settings,
 };
 use crate::{
-    system::{ExecutorPtr, Publisher, PublisherPtr, Subscription},
+    system::{ExecutorPtr, Publisher, PublisherPtr, StoppableTask, StoppableTaskPtr, Subscription},
     util::{logger::verbose, path::expand_path},
-    Result,
+    Error, Result,
 };
 
 #[cfg(target_family = "unix")]
@@ -56,6 +57,95 @@ use smol::fs::unix::PermissionsExt;
 /// Atomic pointer to the p2p interface
 pub type P2pPtr = Arc<P2p>;
 
+/// Maximum number of detached broadcasts retained by a P2P instance.
+/// Calls beyond this limit are rejected with [`Error::BroadcastLimitReached`]
+/// rather than queued, keeping retained payloads and channel lists bounded.
+pub const MAX_CONCURRENT_BROADCASTS: usize = 64;
+
+struct BroadcastTaskState {
+    accepting: bool,
+    tasks: HashSet<StoppableTaskPtr>,
+}
+
+struct BroadcastTasks {
+    state: Mutex<BroadcastTaskState>,
+    rejected: AtomicUsize,
+}
+
+impl BroadcastTasks {
+    fn new() -> Arc<Self> {
+        Arc::new(Self {
+            state: Mutex::new(BroadcastTaskState { accepting: true, tasks: HashSet::new() }),
+            rejected: AtomicUsize::new(0),
+        })
+    }
+
+    fn start<'a, MainFut>(
+        self: &Arc<Self>,
+        main: MainFut,
+        ex: Arc<smol::Executor<'a>>,
+    ) -> Result<()>
+    where
+        MainFut: Future<Output = Result<()>> + Send + 'a,
+    {
+        let task = StoppableTask::new();
+        let mut state = self.state.lock();
+        if !state.accepting {
+            return Err(Error::NetworkServiceStopped)
+        }
+        if state.tasks.len() >= MAX_CONCURRENT_BROADCASTS {
+            self.rejected.fetch_add(1, Ordering::Relaxed);
+            return Err(Error::BroadcastLimitReached)
+        }
+
+        state.tasks.insert(task.clone());
+        let tasks = self.clone();
+        let task_ = task.clone();
+        task.start(
+            main,
+            move |_| async move {
+                tasks.state.lock().tasks.remove(&task_);
+            },
+            Error::DetachedTaskStopped,
+            ex,
+        );
+        Ok(())
+    }
+
+    fn close(&self) -> Vec<StoppableTaskPtr> {
+        let mut state = self.state.lock();
+        state.accepting = false;
+        state.tasks.iter().cloned().collect()
+    }
+
+    fn stop_all_nowait(&self) {
+        for task in self.close() {
+            task.stop_nowait();
+        }
+    }
+
+    async fn stop_all(&self) {
+        for task in self.close() {
+            task.stop().await;
+        }
+        debug_assert!(self.state.lock().tasks.is_empty());
+    }
+
+    fn reopen(&self) {
+        let mut state = self.state.lock();
+        debug_assert!(state.tasks.is_empty());
+        state.accepting = true;
+    }
+
+    fn active(&self) -> usize {
+        self.state.lock().tasks.len()
+    }
+
+    fn rejected(&self) -> usize {
+        self.rejected.load(Ordering::Relaxed)
+    }
+}
+
 /// Toplevel peer-to-peer networking interface
 pub struct P2p {
     /// Global multithreaded executor reference
@@ -86,6 +176,8 @@ pub struct P2p {
     stopping: AtomicBool,
     /// All started channels, including those still performing their handshake.
     channels: Mutex<HashMap<u32, Weak<Channel>>>,
+    /// Bounded set of detached broadcast tasks owned by this P2P instance.
+    broadcast_tasks: Arc<BroadcastTasks>,
 }
 
 impl P2p {
@@ -128,6 +220,7 @@ impl P2p {
             dnet_publisher: Publisher::new(),
             stopping: AtomicBool::new(false),
             channels: Mutex::new(HashMap::new()),
+            broadcast_tasks: BroadcastTasks::new(),
         });
 
         register_default_protocols(self_.clone()).await;
@@ -137,6 +230,7 @@ impl P2p {
 
     /// Starts inbound, outbound, and manual sessions.
     pub async fn start(self: Arc<Self>) -> Result<()> {
+        self.broadcast_tasks.reopen();
         self.stopping.store(false, Ordering::SeqCst);
 
         debug!(target: "net::p2p::start", "P2P::start() [BEGIN] [magic_bytes={:?}]",
@@ -183,6 +277,10 @@ impl P2p {
     pub async fn stop(&self) {
         self.stopping.store(true, Ordering::SeqCst);
 
+        // Reject new broadcasts and cancel retained payloads/channel pointers
+        // before stopping their channels.
+        self.broadcast_tasks.stop_all().await;
+
         // Stop connection producers before draining established channels.
         self.session_inbound().stop().await;
         self.session_manual().stop().await;
@@ -226,13 +324,17 @@ impl P2p {
     }
 
     /// Broadcasts a message concurrently across all active peers.
-    pub async fn broadcast<M: Message>(&self, message: &M) {
+    pub async fn broadcast<M: Message>(&self, message: &M) -> Result<()> {
         self.broadcast_with_exclude(message, &[]).await
     }
 
     /// Broadcasts a message concurrently across active peers, excluding
     /// the ones provided in `exclude_list`.
-    pub async fn broadcast_with_exclude<M: Message>(&self, message: &M, exclude_list: &[Url]) {
+    pub async fn broadcast_with_exclude<M: Message>(
+        &self,
+        message: &M,
+        exclude_list: &[Url],
+    ) -> Result<()> {
         let mut channels = Vec::new();
         for channel in self.hosts().peers() {
             if exclude_list.contains(channel.address()) {
@@ -244,18 +346,49 @@ impl P2p {
     }
 
     /// Broadcast a message concurrently to all given peers.
-    pub async fn broadcast_to<M: Message>(&self, message: &M, channel_list: &[ChannelPtr]) {
+    ///
+    /// The send runs in the background when admitted. If
+    /// [`MAX_CONCURRENT_BROADCASTS`] sends are already active, this returns
+    /// [`Error::BroadcastLimitReached`] without retaining the message or
+    /// channel list. Calls made during shutdown return
+    /// [`Error::NetworkServiceStopped`].
+    pub async fn broadcast_to<M: Message>(
+        &self,
+        message: &M,
+        channel_list: &[ChannelPtr],
+    ) -> Result<()> {
+        if self.is_stopping() {
+            return Err(Error::NetworkServiceStopped)
+        }
+
         if channel_list.is_empty() {
             verbose!(target: "net::p2p::broadcast", "[P2P] No connected channels found for broadcast");
-            return
+            return Ok(())
         }
 
         // Serialize the provided message
         let message = SerializedMessage::new(message).await;
 
-        // Spawn a detached task to actually send the message to the channels,
-        // so we don't block wiating channels that are rate limited.
-        self.executor.spawn(broadcast_serialized_to::<M>(message, channel_list.to_vec())).detach();
+        // Keep rate-limited sends detached while bounding and tracking every
+        // task so shutdown can cancel and drain them.
+        let channels = channel_list.to_vec();
+        self.broadcast_tasks.start(
+            async move {
+                broadcast_serialized_to::<M>(message, channels).await;
+                Ok(())
+            },
+            self.executor.clone(),
+        )
+    }
+
+    /// Number of broadcasts currently retained by this P2P instance.
+    pub fn active_broadcasts(&self) -> usize {
+        self.broadcast_tasks.active()
+    }
+
+    /// Number of broadcasts rejected because the concurrency limit was full.
+    pub fn rejected_broadcasts(&self) -> usize {
+        self.broadcast_tasks.rejected()
     }
 
     /// Check whether this node has connections to any peers. This method will
@@ -368,6 +501,12 @@ impl P2p {
     }
 }
 
+impl Drop for P2p {
+    fn drop(&mut self) {
+        self.broadcast_tasks.stop_all_nowait();
+    }
+}
+
 /// Auxiliary function to broadcast a serialized message concurrently to all given peers.
 async fn broadcast_serialized_to<M: Message>(
     message: SerializedMessage,

src/net/protocol/protocol_generic.rs

@@ -268,7 +268,13 @@ impl<M: Message + Clone, R: Message + Clone + Debug> ProtocolGeneric<M, R> {
             // Handle action signal
             match action {
                 ProtocolGenericAction::Broadcast => {
-                    self.p2p.broadcast_with_exclude(&msg_copy, &exclude_list).await
+                    if let Err(e) = self.p2p.broadcast_with_exclude(&msg_copy, &exclude_list).await
+                    {
+                        debug!(
+                            target: "net::protocol_generic::handle_receive_message",
+                            "Broadcast was not admitted: {e}"
+                        );
+                    }
                 }
                 ProtocolGenericAction::Response(r) => {
                     if let Err(e) = self.channel.send(&r).await {

src/net/session/direct_session.rs

@@ -35,7 +35,7 @@ use std::{
 
 use async_trait::async_trait;
 use smol::lock::{Mutex as AsyncMutex, OnceCell};
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 use url::Url;
 
 use super::{
@@ -598,7 +598,12 @@ impl PeerDiscovery {
                 let get_addrs =
                     GetAddrsMessage { max: getaddrs_max.unwrap_or(1), transports: active_profiles };
 
-                self.p2p().broadcast(&get_addrs).await;
+                if let Err(e) = self.p2p().broadcast(&get_addrs).await {
+                    debug!(
+                        target: "net::direct_session::peer_discovery",
+                        "GetAddrs broadcast was not admitted: {e}"
+                    );
+                }
 
                 // Wait for a hosts store update event
                 let store_sub = self.p2p().hosts().subscribe_store().await;

src/net/session/outbound_session.rs

@@ -661,7 +661,12 @@ impl PeerDiscoveryBase for PeerDiscovery {
                     transports: active_profiles,
                 };
 
-                self.p2p().broadcast(&get_addrs).await;
+                if let Err(e) = self.p2p().broadcast(&get_addrs).await {
+                    debug!(
+                        target: "net::outbound_session::peer_discovery",
+                        "GetAddrs broadcast was not admitted: {e}"
+                    );
+                }
 
                 // Wait for a hosts store update event
                 let store_sub = self.p2p().hosts().subscribe_store().await;

src/net/tests.rs

@@ -37,11 +37,13 @@ use crate::{
         hosts::HostColor,
         message::{GetAddrsMessage, Message},
         metering::{MeteringConfiguration, DEFAULT_METERING_CONFIGURATION},
+        p2p::MAX_CONCURRENT_BROADCASTS,
         settings::NetworkProfile,
         P2p, Settings,
     },
     system::{sleep, timeout::timeout},
     util::logger::{setup_test_logger, Level},
+    Error,
 };
 
 fn init_logger() {
@@ -724,3 +726,90 @@ async fn p2p_shutdown_drains_channels_across_restarts_real(ex: Arc<Executor<'sta
         }
     }
 }
+
+#[test]
+fn p2p_broadcasts_are_bounded_and_drained() {
+    test_body!(p2p_broadcasts_are_bounded_and_drained_real, 2);
+}
+
+async fn p2p_broadcasts_are_bounded_and_drained_real(ex: Arc<Executor<'static>>) {
+    const REJECTED_BROADCASTS: usize = 1024;
+
+    #[derive(SerialEncodable, SerialDecodable)]
+    struct SlowMessage(u32);
+    crate::impl_p2p_message!(
+        SlowMessage,
+        "slow",
+        4,
+        1,
+        MeteringConfiguration {
+            threshold: 1,
+            sleep_step: 60_000,
+            expiry_time: crate::util::time::NanoTimestamp(0),
+        }
+    );
+
+    let port = get_random_available_port();
+    let listen_url = Url::parse(&format!("tcp://127.0.0.1:{port}")).unwrap();
+    let server_settings = Settings {
+        localnet: true,
+        inbound_addrs: vec![listen_url.clone()],
+        inbound_connections: 1,
+        outbound_connections: 0,
+        active_profiles: vec!["tcp".to_string()],
+        ..Default::default()
+    };
+    let client_settings = Settings {
+        localnet: true,
+        peers: vec![listen_url],
+        inbound_connections: 0,
+        outbound_connections: 0,
+        active_profiles: vec!["tcp".to_string()],
+        ..Default::default()
+    };
+
+    let server = P2p::new(server_settings, ex.clone()).await.unwrap();
+    let client = P2p::new(client_settings, ex).await.unwrap();
+    server.clone().start().await.unwrap();
+    client.clone().start().await.unwrap();
+
+    timeout(Duration::from_secs(5), async {
+        while client.hosts().channels().is_empty() || server.hosts().channels().is_empty() {
+            Timer::after(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("manual connection was not established");
+
+    let channel = client.hosts().channels().first().unwrap().clone();
+    let receiving_channel = server.hosts().channels().first().unwrap().clone();
+    receiving_channel.message_subsystem().add_dispatch::<SlowMessage>().await;
+    let channels = vec![channel.clone()];
+
+    // Prime metering so every admitted broadcast waits for two minutes.
+    // This models a slow peer without making the test itself wait.
+    channel.send(&SlowMessage(0)).await.unwrap();
+
+    for i in 0..MAX_CONCURRENT_BROADCASTS {
+        client.broadcast_to(&SlowMessage(i as u32), &channels).await.unwrap();
+    }
+    assert_eq!(client.active_broadcasts(), MAX_CONCURRENT_BROADCASTS);
+
+    for i in 0..REJECTED_BROADCASTS {
+        assert!(matches!(
+            client.broadcast_to(&SlowMessage(i as u32), &channels).await,
+            Err(Error::BroadcastLimitReached)
+        ));
+    }
+    assert_eq!(client.active_broadcasts(), MAX_CONCURRENT_BROADCASTS);
+    assert_eq!(client.rejected_broadcasts(), REJECTED_BROADCASTS);
+
+    client.stop().await;
+    assert_eq!(client.active_broadcasts(), 0);
+    assert!(matches!(
+        client.broadcast_to(&SlowMessage(0), &channels).await,
+        Err(Error::NetworkServiceStopped)
+    ));
+
+    server.stop().await;
+}