feat: set stage resets to 0 at construction

crates/derive/src/macros.rs

@@ -54,4 +54,8 @@ macro_rules! set {
         #[cfg(feature = "metrics")]
         $crate::metrics::$metric.set($value);
     };
+    ($metric:ident, $value:expr, $labels:expr) => {
+        #[cfg(feature = "metrics")]
+        $crate::metrics::$metric.with_label_values($labels).set($value as f64);
+    };
 }

crates/derive/src/stages/attributes_queue.rs

@@ -67,6 +67,7 @@ where
 {
     /// Create a new [AttributesQueue] stage.
     pub fn new(cfg: Arc<RollupConfig>, prev: P, builder: AB) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["attributes-queue"]);
         Self { cfg, prev, is_last_in_span: false, batch: None, builder }
     }
 

crates/derive/src/stages/batch_queue.rs

@@ -80,6 +80,7 @@ where
 {
     /// Creates a new [BatchQueue] stage.
     pub fn new(cfg: Arc<RollupConfig>, prev: P, fetcher: BF) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["batch-queue"]);
         Self {
             cfg,
             prev,

crates/derive/src/stages/channel_bank.rs

@@ -55,6 +55,7 @@ where
 {
     /// Create a new [ChannelBank] stage.
     pub fn new(cfg: Arc<RollupConfig>, prev: P) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["channel-bank"]);
         Self { cfg, channels: HashMap::new(), channel_queue: VecDeque::new(), prev }
     }
 

crates/derive/src/stages/channel_reader.rs

@@ -53,6 +53,7 @@ where
 {
     /// Create a new [ChannelReader] stage.
     pub fn new(prev: P, cfg: Arc<RollupConfig>) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["channel-reader"]);
         Self { prev, next_batch: None, cfg: cfg.clone() }
     }
 

crates/derive/src/stages/frame_queue.rs

@@ -47,6 +47,7 @@ where
     ///
     /// [L1Retrieval]: crate::stages::L1Retrieval
     pub fn new(prev: P) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["frame-queue"]);
         Self { prev, queue: VecDeque::new() }
     }
 }

crates/derive/src/stages/l1_retrieval.rs

@@ -60,6 +60,7 @@ where
     ///
     /// [L1Traversal]: crate::stages::L1Traversal
     pub fn new(prev: P, provider: DAP) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["l1-retrieval"]);
         Self { prev, provider, data: None }
     }
 }

crates/derive/src/stages/l1_traversal.rs

@@ -50,6 +50,7 @@ impl<F: ChainProvider + Send> L1RetrievalProvider for L1Traversal<F> {
 impl<F: ChainProvider> L1Traversal<F> {
     /// Creates a new [L1Traversal] instance.
     pub fn new(data_source: F, cfg: Arc<RollupConfig>) -> Self {
+        crate::set!(STAGE_RESETS, 0, &["l1-traversal"]);
         Self {
             block: Some(BlockInfo::default()),
             data_source,