feat: blob fee horizon and last finalization interval metric (#176)

Co-authored-by: Ahmed Sagdati <[email protected]>

packages/adapters/eth/src/websocket/connection.rs

@@ -81,9 +81,12 @@ pub struct WsConnection {
     metrics: Metrics,
 }
 
+const MAX_BLOB_FEE_HORIZON: u32 = 5;
+
 impl WsConnection {
-    async fn get_next_blob_fee(&self, provider: &WsProvider) -> Result<u128> {
-        provider
+    async fn get_next_blob_fee(&self, horizon: u32) -> Result<u128> {
+        let mut next_block_blob_fee = self
+            .provider
             .get_block_by_number(BlockNumberOrTag::Latest, false)
             .await?
             .ok_or(Error::Network(
@@ -93,15 +96,21 @@ impl WsConnection {
             .next_block_blob_fee()
             .ok_or(Error::Network(
                 "next_block_blob_fee returned None".to_string(),
-            ))
+            ))?;
+
+        for _ in 0..horizon {
+            // multiply by 1.125 = multiply by 9, then divide by 8
+            next_block_blob_fee = next_block_blob_fee.saturating_mul(9).saturating_div(8);
+        }
+        Ok(next_block_blob_fee)
     }
 
     async fn get_bumped_fees(
         &self,
         previous_tx: &L1Tx,
         provider: &WsProvider,
     ) -> Result<(u128, u128, u128)> {
-        let next_blob_fee = self.get_next_blob_fee(provider).await?;
+        let next_blob_fee = self.get_next_blob_fee(MAX_BLOB_FEE_HORIZON).await?;
         let max_fee_per_blob_gas = max(next_blob_fee, previous_tx.blob_fee.saturating_mul(2));
 
         let Eip1559Estimation {
@@ -277,9 +286,14 @@ impl EthApi for WsConnection {
                     .with_blob_sidecar(sidecar)
                     .with_to(*blob_signer_address)
             }
-            _ => TransactionRequest::default()
-                .with_blob_sidecar(sidecar)
-                .with_to(*blob_signer_address),
+            _ => {
+                let blob_fee = self.get_next_blob_fee(MAX_BLOB_FEE_HORIZON).await?;
+
+                TransactionRequest::default()
+                    .with_blob_sidecar(sidecar)
+                    .with_max_fee_per_blob_gas(blob_fee)
+                    .with_to(*blob_signer_address)
+            }
         };
 
         let blob_tx = blob_provider.fill(blob_tx).await?;

packages/adapters/storage/src/lib.rs

@@ -38,6 +38,12 @@ impl services::state_listener::port::Storage for Postgres {
     async fn has_pending_txs(&self) -> Result<bool> {
         self._has_pending_txs().await.map_err(Into::into)
     }
+
+    async fn earliest_submission_attempt(&self, nonce: u32) -> Result<Option<DateTime<Utc>>> {
+        self._earliest_submission_attempt(nonce)
+            .await
+            .map_err(Into::into)
+    }
 }
 
 impl services::cost_reporter::port::Storage for Postgres {

packages/adapters/storage/src/postgres.rs

@@ -448,6 +448,27 @@ impl Postgres {
         Ok(response)
     }
 
+    pub(crate) async fn _earliest_submission_attempt(
+        &self,
+        nonce: u32,
+    ) -> Result<Option<DateTime<Utc>>> {
+        let response = sqlx::query!(
+            r#"SELECT
+            MIN(l1_blob_transaction.created_at) AS earliest_tx_time
+        FROM
+            l1_blob_transaction
+        WHERE
+            l1_blob_transaction.nonce = $1;
+        "#,
+            nonce as i64
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?
+        .and_then(|response| response.earliest_tx_time);
+
+        Ok(response)
+    }
+
     pub(crate) async fn _lowest_unbundled_blocks(
         &self,
         starting_height: u32,

packages/adapters/storage/src/test_instance.rs

@@ -200,6 +200,16 @@ impl services::state_listener::port::Storage for DbWithProcess {
     async fn has_pending_txs(&self) -> services::Result<bool> {
         self.db._has_pending_txs().await.map_err(Into::into)
     }
+
+    async fn earliest_submission_attempt(
+        &self,
+        nonce: u32,
+    ) -> services::Result<Option<DateTime<Utc>>> {
+        self.db
+            ._earliest_submission_attempt(nonce)
+            .await
+            .map_err(Into::into)
+    }
 }
 
 impl block_importer::port::Storage for DbWithProcess {

packages/services/src/state_listener.rs

@@ -135,6 +135,17 @@ pub mod service {
 
                 info!("blob tx {} finalized", hex::encode(tx.hash));
 
+                let earliest_submission_attempt =
+                    self.storage.earliest_submission_attempt(tx.nonce).await?;
+
+                self.metrics.last_finalization_interval.set(
+                    earliest_submission_attempt
+                        .map(|earliest_submission_attempt| {
+                            (now - earliest_submission_attempt).num_seconds()
+                        })
+                        .unwrap_or(0),
+                );
+
                 self.metrics
                     .last_eth_block_w_blob
                     .set(i64::try_from(tx_response.block_number()).unwrap_or(i64::MAX))
@@ -177,13 +188,15 @@ pub mod service {
     struct Metrics {
         last_eth_block_w_blob: IntGauge,
         last_finalization_time: IntGauge,
+        last_finalization_interval: IntGauge,
     }
 
     impl<L1, Db, Clock> RegistersMetrics for StateListener<L1, 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()),
             ]
         }
     }
@@ -196,9 +209,18 @@ pub mod service {
             ))
             .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",
+            )
+            .expect(
+                "seconds_from_earliest_submission_to_finalization gauge to be correctly configured",
+            );
+
             Self {
                 last_eth_block_w_blob,
                 last_finalization_time,
+                last_finalization_interval,
             }
         }
     }
@@ -240,6 +262,7 @@ pub mod port {
             cost_per_tx: Vec<TransactionCostUpdate>,
         ) -> Result<()>;
         async fn has_pending_txs(&self) -> Result<bool>;
+        async fn earliest_submission_attempt(&self, nonce: u32) -> Result<Option<DateTime<Utc>>>;
     }
 
     pub trait Clock {