feat(derive): track the current channel size

crates/derive/src/macros.rs

@@ -20,6 +20,10 @@ macro_rules! timer {
 /// Increments a metric with a label value.
 #[macro_export]
 macro_rules! inc {
+    ($metric:ident) => {
+        #[cfg(feature = "metrics")]
+        $crate::metrics::$metric.inc();
+    };
     ($metric:ident, $labels:expr) => {
         #[cfg(feature = "metrics")]
         $crate::metrics::$metric.with_label_values($labels).inc();

crates/derive/src/metrics.rs

@@ -20,6 +20,12 @@ lazy_static! {
         "Tracks the L1 origin for the L1 Traversal Stage"
     ).expect("Origin Gauge failed to register");
 
+    /// Tracks the number of frames in the current channel.
+    pub static ref CURRENT_CHANNEL_FRAMES: IntGauge = register_int_gauge!(
+        "kona_derive_current_channel_frames",
+        "Tracks the number of frames in the current channel."
+    ).expect("Current channel frames failed to register");
+
     /// Tracks batch reader errors.
     pub static ref BATCH_READER_ERRORS: CounterVec = register_counter_vec!(
         "kona_derive_batch_reader_errors",

crates/derive/src/stages/channel_bank.rs

@@ -98,6 +98,7 @@ where
             warn!(target: "channel-bank", "Failed to add frame to channel: {:?}", frame_id);
             return Ok(());
         }
+        crate::inc!(CURRENT_CHANNEL_FRAMES);
 
         self.prune()
     }
@@ -122,6 +123,14 @@ where
             crate::observe!(CHANNEL_TIMEOUTS, (origin.number - channel.open_block_number()) as f64);
             self.channels.remove(&first);
             self.channel_queue.pop_front();
+            crate::set!(
+                CURRENT_CHANNEL_FRAMES,
+                self.channel_queue.front().map_or(0, |id| self
+                    .channels
+                    .get(id)
+                    .map_or(0, |c| c.len())
+                    as i64)
+            );
             return Ok(None);
         }
 

crates/derive/src/types/channel.rs

@@ -40,6 +40,16 @@ impl Channel {
         Self { id, open_block, inputs: HashMap::new(), ..Default::default() }
     }
 
+    /// Returns the number of frames ingested.
+    pub fn len(&self) -> usize {
+        self.inputs.len()
+    }
+
+    /// Returns if the channel is empty.
+    pub fn is_empty(&self) -> bool {
+        self.inputs.is_empty()
+    }
+
     /// Add a frame to the channel.
     ///
     /// ## Takes