notifications: Exit protocols on handle drop to save up CPU of minimal-relay-chains (#376)

This PR ensures that notification tasks are fully terminated when the
associated handle is dropped, addressing an issue specific to
`minimal-relay-chains`. We have operated from the assumption that the
handle is never supposed to get dropped, since protocols cannot be
dynamically injected after the backend is started.

This PR effectively reduces the CPU usage of minimal-relay-chains from
7% to 0.1%.

We have noticed that the CPU consumption of minimal-relay-chains has
increased compared to libp2p backends. However, our metrics for CPU
consumption of full relay chains have drastically improved.

The issue is the following:
- substrate needs a block-announces notification config to initiate the
network components.
- the minimal-relay-chains do not need this notification around, and
drop the associated handle.
- this behaviour causes a CPU loop that continuously spams the following
log line:

```
litep2p::notification: [Parachain] user protocol has exited, exiting protocol=/b0a8d493285c2df73290dfb7e61f870f17b41801197a149ca93654499ea3dafe/block-announces/1
```

CPU before and after:
![Screenshot 2025-04-17 at 17 44
23](https://github.com/user-attachments/assets/34a91cd4-7354-498a-ae99-b77d7b103f5b)

Fixes: paritytech/polkadot-sdk#7998

---------

Signed-off-by: Alexandru Vasile <[email protected]>
Co-authored-by: Bastian Köcher <[email protected]>

Author: lexnv

src/protocol/notification/mod.rs

@@ -1606,7 +1606,9 @@ impl NotificationProtocol {
     }
 
     /// Handle next notification event.
-    async fn next_event(&mut self) {
+    ///
+    /// Returns `true` when the user command stream was dropped.
+    async fn next_event(&mut self) -> bool {
         // biased select is used because the substream events must be prioritized above other events
         // that is because a closed substream is detected by either `substreams` or `negotiation`
         // and if that event is not handled with priority but, e.g., inbound substream is
@@ -1783,9 +1785,16 @@ impl NotificationProtocol {
                     );
                 }
             }
+
+            // User commands.
             command = self.command_rx.recv() => match command {
                 None => {
-                    tracing::debug!(target: LOG_TARGET, "user protocol has exited, exiting");
+                    tracing::debug!(
+                        target: LOG_TARGET,
+                        protocol = %self.protocol,
+                        "user protocol has exited, exiting"
+                    );
+                    return true;
                 }
                 Some(command) => match command {
                     NotificationCommand::OpenSubstream { peers } => {
@@ -1811,14 +1820,14 @@ impl NotificationProtocol {
                 }
             },
         }
+
+        false
     }
 
     /// Start [`NotificationProtocol`] event loop.
     pub(crate) async fn run(mut self) {
         tracing::debug!(target: LOG_TARGET, "starting notification event loop");
 
-        loop {
-            self.next_event().await;
-        }
+        while !self.next_event().await {}
     }
 }

src/protocol/notification/tests/notification.rs

@@ -1117,3 +1117,22 @@ async fn second_inbound_substream_opened_while_outbound_substream_was_opening()
         state => panic!("invalid state for peer: {state:?}"),
     }
 }
+
+#[tokio::test]
+async fn drop_handle_exits_protocol() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .try_init();
+
+    let (mut protocol, handle, _sender, _tx) = make_notification_protocol();
+
+    // Simulate a handle drop.
+    drop(handle);
+
+    // Call `next_event` and ensure it returns true.
+    let result = protocol.next_event().await;
+    assert!(
+        result,
+        "Expected `next_event` to return true when `command_rx` is dropped"
+    );
+}