chore(gas_price_service_v1): strictly ensure last_recorded_height is set, to avoid initial poll of da source

crates/services/gas_price_service/src/common/fuel_core_storage_adapter.rs

@@ -119,7 +119,7 @@ impl<Storage> GasPriceServiceAtomicStorage for Storage
 where
     Storage: 'static,
     Storage: GetMetadataStorage + GetLatestRecordedHeight,
-    Storage: KeyValueInspect<Column = GasPriceColumn> + Modifiable + Send + Sync,
+    Storage: KeyValueInspect<Column = GasPriceColumn> + Modifiable + Send + Sync + Clone,
 {
     type Transaction<'a> = StorageTransaction<&'a mut Storage> where Self: 'a;
 

crates/services/gas_price_service/src/v1/da_source_service.rs

@@ -60,10 +60,12 @@ mod tests {
         let da_block_costs_source =
             DummyDaBlockCosts::new(Ok(expected_da_cost.clone()), notifier.clone());
         let latest_l2_height = Arc::new(Mutex::new(BlockHeight::new(10u32)));
+        let recorded_height = BlockHeight::new(0);
         let service = new_da_service(
             da_block_costs_source,
             Some(Duration::from_millis(1)),
             latest_l2_height,
+            recorded_height,
         );
         let mut shared_state = &mut service.shared.subscribe();
 
@@ -84,10 +86,12 @@ mod tests {
         let da_block_costs_source =
             DummyDaBlockCosts::new(Err(anyhow::anyhow!("boo!")), notifier.clone());
         let latest_l2_height = latest_l2_height(0);
+        let recorded_height = BlockHeight::new(0);
         let service = new_da_service(
             da_block_costs_source,
             Some(Duration::from_millis(1)),
             latest_l2_height,
+            recorded_height,
         );
         let mut shared_state = &mut service.shared.subscribe();
 
@@ -117,10 +121,12 @@ mod tests {
         let da_block_costs_source =
             DummyDaBlockCosts::new(Ok(unexpected_costs.clone()), notifier.clone());
         let latest_l2_height = latest_l2_height(l2_height);
+        let recorded_height = BlockHeight::new(0);
         let service = new_da_service(
             da_block_costs_source,
             Some(Duration::from_millis(1)),
             latest_l2_height,
+            recorded_height,
         );
         let mut shared_state = &mut service.shared.subscribe();
 
@@ -153,11 +159,12 @@ mod tests {
         let da_block_costs_source =
             DummyDaBlockCosts::new(Ok(unexpected_costs.clone()), notifier.clone());
         let latest_l2_height = latest_l2_height(l2_height);
+        let recorded_height = BlockHeight::new(0);
         let mut service = DaSourceService::new(
             da_block_costs_source,
             Some(Duration::from_millis(1)),
             latest_l2_height,
-            None,
+            recorded_height,
         );
         let mut watcher = StateWatcher::started();
 
@@ -166,8 +173,7 @@ mod tests {
 
         // then
         let recorded_height = service.recorded_height();
-        let expected = 1;
-        assert!(recorded_height.is_none())
+        assert_eq!(*recorded_height, 0);
     }
 
     #[tokio::test]
@@ -191,11 +197,12 @@ mod tests {
         let latest_l2_height = latest_l2_height(l2_height);
         let (sender, mut receiver) =
             tokio::sync::broadcast::channel(DA_BLOCK_COSTS_CHANNEL_SIZE);
+        let expected_height = BlockHeight::new(recorded_height);
         let mut service = DaSourceService::new_with_sender(
             da_block_costs_source,
             Some(Duration::from_millis(1)),
             latest_l2_height,
-            None,
+            expected_height,
             sender,
         );
         let mut watcher = StateWatcher::started();
@@ -204,8 +211,7 @@ mod tests {
         let next = service.run(&mut watcher).await;
 
         // then
-        let actual = service.recorded_height().unwrap();
-        let expected = BlockHeight::from(recorded_height);
-        assert_eq!(expected, actual);
+        let actual = service.recorded_height();
+        assert_eq!(expected_height, actual);
     }
 }

crates/services/gas_price_service/src/v1/da_source_service/block_committer_costs.rs

@@ -20,8 +20,6 @@ use std::ops::Deref;
 
 #[async_trait::async_trait]
 pub trait BlockCommitterApi: Send + Sync {
-    /// Used on first run to get the latest costs and seqno
-    async fn get_latest_costs(&self) -> DaBlockCostsResult<Option<RawDaBlockCosts>>;
     /// Used to get the costs for a specific seqno
     async fn get_costs_by_l2_block_number(
         &self,
@@ -90,17 +88,17 @@ where
 {
     async fn request_da_block_costs(
         &mut self,
-        last_recorded_height: &Option<BlockHeight>,
+        last_recorded_height: &BlockHeight,
     ) -> DaBlockCostsResult<Vec<DaBlockCosts>> {
-        let raw_da_block_costs: Vec<_> = match last_recorded_height.and_then(|x| x.succ())
-        {
-            Some(ref next_height) => {
-                self.client
-                    .get_costs_by_l2_block_number(*next_height.deref())
-                    .await?
-            }
-            None => self.client.get_latest_costs().await?.into_iter().collect(),
-        };
+        let next_height = last_recorded_height.succ().ok_or(anyhow!(
+            "Failed to increment the last recorded height: {:?}",
+            last_recorded_height
+        ))?;
+
+        let raw_da_block_costs: Vec<_> = self
+            .client
+            .get_costs_by_l2_block_number(*next_height.deref())
+            .await?;
 
         let da_block_costs: Vec<_> =
             raw_da_block_costs.iter().map(DaBlockCosts::from).collect();
@@ -141,19 +139,6 @@ impl BlockCommitterApi for BlockCommitterHttpApi {
             Ok(vec![])
         }
     }
-
-    async fn get_latest_costs(&self) -> DaBlockCostsResult<Option<RawDaBlockCosts>> {
-        // Latest: http://localhost:8080/v1/costs?variant=latest&limit=5
-        if let Some(url) = &self.url {
-            let formatted_url = format!("{url}/v1/costs?variant=latest&limit=1");
-            let response = self.client.get(formatted_url).send().await?;
-            let raw_da_block_costs = response.json::<Vec<RawDaBlockCosts>>().await?;
-            // only take the first element, since we are only looking for the most recent
-            Ok(raw_da_block_costs.first().cloned())
-        } else {
-            Ok(None)
-        }
-    }
 }
 
 #[cfg(test)]
@@ -269,56 +254,6 @@ mod test_block_committer_http_api {
         // then
         assert_eq!(actual, expected);
     }
-
-    #[test]
-    fn get_latest_costs__when_url_is_none__then_returns_none() {
-        let rt = tokio::runtime::Runtime::new().unwrap();
-
-        // given
-        let block_committer = BlockCommitterHttpApi::new(None);
-
-        // when
-        let actual =
-            rt.block_on(async { block_committer.get_latest_costs().await.unwrap() });
-
-        // then
-        assert_eq!(actual, None);
-    }
-
-    #[test]
-    fn get_latest_costs__when_url_is_some__then_returns_expected_costs() {
-        let rt = tokio::runtime::Runtime::new().unwrap();
-        let mut mock = FakeServer::new();
-        let url = mock.url();
-
-        // given
-        let block_committer = BlockCommitterHttpApi::new(Some(url));
-        let not_expected = RawDaBlockCosts {
-            id: 1,
-            start_height: 1,
-            end_height: 10,
-            da_block_height: 1u64.into(),
-            cost: 1,
-            size: 1,
-        };
-        mock.add_response(not_expected);
-        let expected = RawDaBlockCosts {
-            id: 2,
-            start_height: 11,
-            end_height: 20,
-            da_block_height: 2u64.into(),
-            cost: 2,
-            size: 2,
-        };
-        mock.add_response(expected.clone());
-
-        // when
-        let actual =
-            rt.block_on(async { block_committer.get_latest_costs().await.unwrap() });
-
-        // then
-        assert_eq!(actual, Some(expected));
-    }
 }
 #[cfg(any(test, feature = "test-helpers"))]
 pub mod fake_server {
@@ -450,9 +385,6 @@ mod tests {
 
     #[async_trait::async_trait]
     impl BlockCommitterApi for MockBlockCommitterApi {
-        async fn get_latest_costs(&self) -> DaBlockCostsResult<Option<RawDaBlockCosts>> {
-            Ok(self.value.clone())
-        }
         async fn get_costs_by_l2_block_number(
             &self,
             l2_block_number: u32,
@@ -482,22 +414,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn request_da_block_cost__when_last_value_is_none__then_get_latest_costs_is_called(
-    ) {
-        // given
-        let da_block_costs = test_da_block_costs();
-        let expected = vec![(&da_block_costs).into()];
-        let mock_api = MockBlockCommitterApi::new(Some(da_block_costs));
-        let mut block_committer = BlockCommitterDaBlockCosts::new(mock_api);
-
-        // when
-        let actual = block_committer.request_da_block_costs(&None).await.unwrap();
-
-        // then
-        assert_eq!(actual, expected);
-    }
-
     #[tokio::test]
     async fn request_da_block_cost__when_last_value_is_some__then_get_costs_by_l2_block_number_is_called(
     ) {
@@ -510,7 +426,7 @@ mod tests {
 
         // when
         let actual = block_committer
-            .request_da_block_costs(&Some(latest_height))
+            .request_da_block_costs(&latest_height)
             .await
             .unwrap();
 

crates/services/gas_price_service/src/v1/da_source_service/dummy_costs.rs

@@ -25,7 +25,7 @@ impl DummyDaBlockCosts {
 impl DaBlockCostsSource for DummyDaBlockCosts {
     async fn request_da_block_costs(
         &mut self,
-        _latest_recorded_height: &Option<BlockHeight>,
+        _latest_recorded_height: &BlockHeight,
     ) -> DaBlockCostsResult<Vec<DaBlockCosts>> {
         match &self.value {
             Ok(da_block_costs) => {

crates/services/gas_price_service/src/v1/da_source_service/service.rs

@@ -44,7 +44,7 @@ pub struct DaSourceService<Source> {
     source: Source,
     shared_state: SharedState,
     latest_l2_height: Arc<Mutex<BlockHeight>>,
-    recorded_height: Option<BlockHeight>,
+    recorded_height: BlockHeight,
 }
 
 pub(crate) const DA_BLOCK_COSTS_CHANNEL_SIZE: usize = 16 * 1024;
@@ -58,7 +58,7 @@ where
         source: Source,
         poll_interval: Option<Duration>,
         latest_l2_height: Arc<Mutex<BlockHeight>>,
-        recorded_height: Option<BlockHeight>,
+        recorded_height: BlockHeight,
     ) -> Self {
         let (sender, _) = tokio::sync::broadcast::channel(DA_BLOCK_COSTS_CHANNEL_SIZE);
         #[allow(clippy::arithmetic_side_effects)]
@@ -78,7 +78,7 @@ where
         source: Source,
         poll_interval: Option<Duration>,
         latest_l2_height: Arc<Mutex<BlockHeight>>,
-        recorded_height: Option<BlockHeight>,
+        recorded_height: BlockHeight,
         sender: Sender<DaBlockCosts>,
     ) -> Self {
         Self {
@@ -109,12 +109,8 @@ where
             tracing::debug!("Sending block costs: {:?}", da_block_costs);
             let end = BlockHeight::from(*da_block_costs.l2_blocks.end());
             self.shared_state.0.send(da_block_costs)?;
-            if let Some(recorded_height) = self.recorded_height {
-                if end > recorded_height {
-                    self.recorded_height = Some(end)
-                }
-            } else {
-                self.recorded_height = Some(end)
+            if end > self.recorded_height {
+                self.recorded_height = end
             }
         }
         Ok(())
@@ -136,7 +132,7 @@ where
     }
 
     #[cfg(test)]
-    pub fn recorded_height(&self) -> Option<BlockHeight> {
+    pub fn recorded_height(&self) -> BlockHeight {
         self.recorded_height
     }
 }
@@ -147,7 +143,7 @@ where
 pub trait DaBlockCostsSource: Send + Sync {
     async fn request_da_block_costs(
         &mut self,
-        recorded_height: &Option<BlockHeight>,
+        recorded_height: &BlockHeight,
     ) -> Result<Vec<DaBlockCosts>>;
 }
 
@@ -210,11 +206,12 @@ pub fn new_da_service<S: DaBlockCostsSource>(
     da_source: S,
     poll_interval: Option<Duration>,
     latest_l2_height: Arc<Mutex<BlockHeight>>,
+    recorded_height: BlockHeight,
 ) -> ServiceRunner<DaSourceService<S>> {
     ServiceRunner::new(DaSourceService::new(
         da_source,
         poll_interval,
         latest_l2_height,
-        None,
+        recorded_height,
     ))
 }