fix: backpressure & stalled protocol generation

chain-signatures/node/src/metrics/messaging.rs

@@ -1,6 +1,6 @@
 use std::sync::LazyLock;
 
-use prometheus::{exponential_buckets, Counter, Histogram, HistogramVec};
+use prometheus::{exponential_buckets, Counter, Histogram, HistogramVec, IntGaugeVec};
 
 use super::{
     try_create_counter_vec_with_node_account_id, try_create_histogram_vec_with_node_account_id,
@@ -78,3 +78,22 @@ pub(crate) static WEB_ENDPOINT_LATENCY: LazyLock<HistogramVec> = LazyLock::new(|
     )
     .unwrap()
 });
+
+pub(crate) static CHANNEL_QUEUE_SIZE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
+    super::try_create_int_gauge_vec_with_node_account_id(
+        "multichain_message_channel_queue_size",
+        "Estimated number of buffered messages queued per message channel",
+        &["channel", "channel_id"],
+    )
+    .unwrap()
+});
+
+pub(crate) fn set_channel_queue_size(channel: &str, channel_id: &str, len: usize) {
+    CHANNEL_QUEUE_SIZE
+        .with_label_values(&[channel, channel_id])
+        .set(len as i64);
+}
+
+pub(crate) fn remove_channel_queue_size(channel: &str, channel_id: &str) {
+    let _ = CHANNEL_QUEUE_SIZE.remove_label_values(&[channel, channel_id]);
+}

chain-signatures/node/src/protocol/message/filter.rs

@@ -1,6 +1,6 @@
 use std::num::NonZeroUsize;
 
-use tokio::sync::mpsc::{self, error::TryRecvError};
+use tokio::sync::mpsc;
 
 use super::types::{MessageFilterId, Protocols};
 
@@ -13,13 +13,18 @@ pub const MAX_FILTER_SIZE: NonZeroUsize = NonZeroUsize::new(64 * 1024).unwrap();
 
 #[derive(Debug)]
 pub(crate) struct MessageFilter {
+    filter_tx: mpsc::Sender<(Protocols, u64)>,
     filter_rx: mpsc::Receiver<(Protocols, u64)>,
     filter: lru::LruCache<(Protocols, u64), ()>,
 }
 
 impl MessageFilter {
-    pub fn new(filter_rx: mpsc::Receiver<(Protocols, u64)>) -> Self {
+    pub fn new(
+        filter_tx: mpsc::Sender<(Protocols, u64)>,
+        filter_rx: mpsc::Receiver<(Protocols, u64)>,
+    ) -> Self {
         Self {
+            filter_tx,
             filter_rx,
             filter: lru::LruCache::new(MAX_FILTER_SIZE),
         }
@@ -37,15 +42,26 @@ impl MessageFilter {
         };
 
         self.filter.put((msg_type, id), ());
+        crate::metrics::messaging::set_channel_queue_size(
+            "filter",
+            "singleton",
+            self.filter_tx.max_capacity() - self.filter_tx.capacity(),
+        );
     }
 
     pub fn try_update(&mut self) {
-        loop {
-            let (msg_type, id) = match self.filter_rx.try_recv() {
-                Ok(filter) => filter,
-                Err(TryRecvError::Empty | TryRecvError::Disconnected) => return,
-            };
+        let mut updated = false;
+        while let Ok((msg_type, id)) = self.filter_rx.try_recv() {
             self.filter.put((msg_type, id), ());
+            updated = true;
+        }
+
+        if updated {
+            crate::metrics::messaging::set_channel_queue_size(
+                "filter",
+                "singleton",
+                self.filter_tx.max_capacity() - self.filter_tx.capacity(),
+            );
         }
     }