fix: add interval for finalization metric

Author: rymnc

.sqlx/query-d6ad3af8502339a76d4a880f25b887858f300b7b36d0490c244e05d3fd20ee4c.json

@@ -15,8 +15,8 @@ use services::{
     block_bundler::port::UnbundledBlocks,
     types::{
         BlockSubmission, BlockSubmissionTx, BundleCost, CompressedFuelBlock, DateTime,
-        DispersalStatus, EigenDASubmission, Fragment, L1Tx, NonEmpty, NonNegative,
-        TransactionCostUpdate, TransactionState, Utc, storage::BundleFragment,
+        DispersalStatus, EigenDARequestId, EigenDASubmission, Fragment, L1Tx, NonEmpty,
+        NonNegative, TransactionCostUpdate, TransactionState, Utc, storage::BundleFragment,
     },
 };
 
@@ -46,6 +46,15 @@ impl services::state_listener::port::Storage for Postgres {
             .map_err(Into::into)
     }
 
+    async fn earliest_eigen_submission_attempt(
+        &self,
+        request_id: &EigenDARequestId,
+    ) -> Result<Option<DateTime<Utc>>> {
+        self._earliest_eigen_submission_attempt(request_id)
+            .await
+            .map_err(Into::into)
+    }
+
     async fn get_non_finalized_eigen_submission(&self) -> services::Result<Vec<EigenDASubmission>> {
         self._get_non_finalized_eigen_submission()
             .await

packages/adapters/storage/src/lib.rs

@@ -7,8 +7,8 @@ use services::{
     block_bundler::port::UnbundledBlocks,
     types::{
         BlockSubmission, BlockSubmissionTx, BundleCost, CompressedFuelBlock, DateTime,
-        DispersalStatus, EigenDASubmission, Fragment, NonEmpty, NonNegative, TransactionCostUpdate,
-        TransactionState, Utc, storage::SequentialFuelBlocks,
+        DispersalStatus, EigenDARequestId, EigenDASubmission, Fragment, NonEmpty, NonNegative,
+        TransactionCostUpdate, TransactionState, Utc, storage::SequentialFuelBlocks,
     },
 };
 use sqlx::{
@@ -1363,6 +1363,27 @@ impl Postgres {
         })
     }
 
+    pub(crate) async fn _earliest_eigen_submission_attempt(
+        &self,
+        request_id: &EigenDARequestId,
+    ) -> Result<Option<DateTime<Utc>>> {
+        let response = sqlx::query!(
+            r#"SELECT
+            MIN(created_at) AS earliest_submission_time
+        FROM
+            eigen_submission
+        WHERE
+            request_id = $1;
+        "#,
+            request_id
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?
+        .and_then(|response| response.earliest_submission_time);
+
+        Ok(response)
+    }
+
     pub(crate) async fn _record_eigenda_submission(
         &self,
         submission: EigenDASubmission,

packages/adapters/storage/src/postgres.rs

@@ -10,8 +10,8 @@ use services::{
     block_committer, block_importer,
     types::{
         BlockSubmission, BlockSubmissionTx, BundleCost, CompressedFuelBlock, DateTime,
-        DispersalStatus, EigenDASubmission, Fragment, L1Tx, NonEmpty, NonNegative,
-        TransactionCostUpdate, TransactionState, Utc, storage::BundleFragment,
+        DispersalStatus, EigenDARequestId, EigenDASubmission, Fragment, L1Tx, NonEmpty,
+        NonNegative, TransactionCostUpdate, TransactionState, Utc, storage::BundleFragment,
     },
 };
 use sqlx::Executor;
@@ -216,6 +216,14 @@ impl services::state_listener::port::Storage for DbWithProcess {
     async fn get_non_finalized_eigen_submission(&self) -> services::Result<Vec<EigenDASubmission>> {
         unimplemented!();
     }
+
+    async fn earliest_eigen_submission_attempt(
+        &self,
+        _request_id: &EigenDARequestId,
+    ) -> services::Result<Option<DateTime<Utc>>> {
+        unimplemented!();
+    }
+
     async fn update_eigen_submissions(
         &self,
         _changes: Vec<(u32, DispersalStatus)>,

packages/adapters/storage/src/test_instance.rs

@@ -1,6 +1,6 @@
 use metrics::{
     RegistersMetrics,
-    prometheus::{IntGauge, Opts, core::Collector},
+    prometheus::{IntGauge, core::Collector},
 };
 
 use crate::{
@@ -42,6 +42,7 @@ where
         non_finalized: Vec<EigenDASubmission>,
     ) -> crate::Result<()> {
         let mut changes = Vec::with_capacity(non_finalized.len());
+        let mut last_finalized_request_id = None;
 
         for submission in non_finalized {
             let status = self
@@ -82,8 +83,7 @@ where
                     changes.push((submission_id, DispersalStatus::Confirmed));
                 }
                 DispersalStatus::Finalized => {
-                    let now = self.clock.now();
-                    self.metrics.last_finalization_time.set(now.timestamp());
+                    last_finalized_request_id = Some(submission.request_id.clone());
                     tracing::info!(
                         "Finalized submission with request_id: {}",
                         submission.as_base64(),
@@ -108,6 +108,22 @@ where
             }
         }
 
+        // get the last request that was finalized, and update metrics accordingly -
+        if let Some(request_id) = last_finalized_request_id {
+            if let Some(earliest_submission_attempt) = self
+                .storage
+                .earliest_eigen_submission_attempt(&request_id)
+                .await?
+            {
+                let now = self.clock.now();
+                let duration = now.signed_duration_since(earliest_submission_attempt);
+                self.metrics
+                    .last_finalization_interval
+                    .set(duration.num_seconds());
+                self.metrics.last_finalization_time.set(now.timestamp());
+            }
+        }
+
         self.storage.update_eigen_submissions(changes).await?;
 
         Ok(())
@@ -123,7 +139,7 @@ where
     async fn run(&mut self) -> crate::Result<()> {
         let non_finalized = self.storage.get_non_finalized_eigen_submission().await?;
 
-        tracing::info!(
+        tracing::debug!(
             "Checking non-finalized submissions: {}",
             non_finalized.len()
         );
@@ -140,15 +156,13 @@ where
 
 #[derive(Clone)]
 struct Metrics {
-    last_eth_block_w_blob: IntGauge,
     last_finalization_time: IntGauge,
     last_finalization_interval: IntGauge,
 }
 
 impl<EigenDA, Db, Clock> RegistersMetrics for StateListener<EigenDA, Db, Clock> {
     fn metrics(&self) -> Vec<Box<dyn Collector>> {
         vec![
-            Box::new(self.metrics.last_eth_block_w_blob.clone()),
             Box::new(self.metrics.last_finalization_time.clone()),
             Box::new(self.metrics.last_finalization_interval.clone()),
         ]
@@ -157,12 +171,6 @@ impl<EigenDA, Db, Clock> RegistersMetrics for StateListener<EigenDA, Db, Clock>
 
 impl Metrics {
     fn new(last_finalization_time: IntGauge) -> Self {
-        let last_eth_block_w_blob = IntGauge::with_opts(Opts::new(
-            "last_eth_block_w_blob",
-            "The height of the latest Ethereum block used for state submission.",
-        ))
-        .expect("last_eth_block_w_blob metric to be correctly configured");
-
         let last_finalization_interval = IntGauge::new(
             "seconds_from_earliest_submission_to_finalization",
             "The number of seconds from the earliest submission to finalization",
@@ -172,7 +180,6 @@ impl Metrics {
         );
 
         Self {
-            last_eth_block_w_blob,
             last_finalization_time,
             last_finalization_interval,
         }

packages/services/src/state_listener/eigen_service.rs

@@ -1,8 +1,8 @@
 use crate::{
     Result,
     types::{
-        DateTime, DispersalStatus, EigenDASubmission, L1Tx, TransactionCostUpdate,
-        TransactionState, Utc,
+        DateTime, DispersalStatus, EigenDARequestId, EigenDASubmission, L1Tx,
+        TransactionCostUpdate, TransactionState, Utc,
     },
 };
 
@@ -51,6 +51,10 @@ pub trait Storage: Sync {
 
     // EigenDA
     async fn get_non_finalized_eigen_submission(&self) -> Result<Vec<EigenDASubmission>>;
+    async fn earliest_eigen_submission_attempt(
+        &self,
+        request_id: &EigenDARequestId,
+    ) -> Result<Option<DateTime<Utc>>>;
     async fn update_eigen_submissions(&self, changes: Vec<(u32, DispersalStatus)>) -> Result<()>;
 }
 

packages/services/src/state_listener/port.rs

@@ -10,10 +10,12 @@ pub enum DispersalStatus {
     Other(String),
 }
 
+pub type EigenDARequestId = Vec<u8>;
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct EigenDASubmission {
     pub id: Option<u64>,
-    pub request_id: Vec<u8>,
+    pub request_id: EigenDARequestId,
     pub created_at: Option<DateTime<Utc>>,
     pub status: DispersalStatus,
 }