feat(gas_price_service_v1): define RunnableTask for GasPriceServiceV1

CHANGELOG.md

@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2362](https://github.com/FuelLabs/fuel-core/pull/2362): Added a new request_response protocol version `/fuel/req_res/0.0.2`. In comparison with `/fuel/req/0.0.1`, which returns an empty response when a request cannot be fulfilled, this version returns more meaningful error codes. Nodes still support the version `0.0.1` of the protocol to guarantee backward compatibility with fuel-core nodes. Empty responses received from nodes using the old protocol `/fuel/req/0.0.1` are automatically converted into an error `ProtocolV1EmptyResponse` with error code 0, which is also the only error code implemented. More specific error codes will be added in the future.
 - [2386](https://github.com/FuelLabs/fuel-core/pull/2386): Add a flag to define the maximum number of file descriptors that RocksDB can use. By default it's half of the OS limit.
 - [2376](https://github.com/FuelLabs/fuel-core/pull/2376): Add a way to fetch transactions in P2P without specifying a peer.
+- [2416](https://github.com/FuelLabs/fuel-core/issues/2416): Define the `GasPriceServiceV1` task.
 
 ### Fixed
 - [2366](https://github.com/FuelLabs/fuel-core/pull/2366): The `importer_gas_price_for_block` metric is properly collected.

crates/fuel-gas-price-algorithm/src/v1.rs

@@ -313,6 +313,7 @@ impl AlgorithmUpdaterV1 {
         if !height_range.is_empty() {
             self.da_block_update(height_range, range_cost)?;
             self.recalculate_projected_cost();
+            self.update_da_gas_price();
         }
         Ok(())
     }

crates/fuel-gas-price-algorithm/src/v1/tests/update_da_record_data_tests.rs

@@ -270,3 +270,166 @@ fn update_da_record_data__da_block_updates_projected_total_cost_with_known_and_g
     let expected = new_known_total_cost + guessed_part;
     assert_eq!(actual, expected as u128);
 }
+
+#[test]
+fn update_da_record_data__da_block_lowers_da_gas_price() {
+    // given
+    let da_cost_per_byte = 40;
+    let da_recorded_block_height = 10;
+    let l2_block_height = 11;
+    let original_known_total_cost = 150;
+    let unrecorded_blocks = vec![BlockBytes {
+        height: 11,
+        block_bytes: 3000,
+    }];
+    let da_p_component = 2;
+    let guessed_cost: u64 = unrecorded_blocks
+        .iter()
+        .map(|block| block.block_bytes * da_cost_per_byte)
+        .sum();
+    let projected_total_cost = original_known_total_cost + guessed_cost;
+
+    let mut updater = UpdaterBuilder::new()
+        .with_da_cost_per_byte(da_cost_per_byte as u128)
+        .with_da_p_component(da_p_component)
+        .with_last_profit(10, 0)
+        .with_da_recorded_block_height(da_recorded_block_height)
+        .with_l2_block_height(l2_block_height)
+        .with_projected_total_cost(projected_total_cost as u128)
+        .with_known_total_cost(original_known_total_cost as u128)
+        .with_unrecorded_blocks(unrecorded_blocks.clone())
+        .build();
+
+    let new_cost_per_byte = 100;
+    let (recorded_heights, recorded_cost) =
+        unrecorded_blocks
+            .iter()
+            .fold((vec![], 0), |(mut range, cost), block| {
+                range.push(block.height);
+                (range, cost + block.block_bytes * new_cost_per_byte)
+            });
+    let min = recorded_heights.iter().min().unwrap();
+    let max = recorded_heights.iter().max().unwrap();
+    let recorded_range = *min..(max + 1);
+
+    let old_da_gas_price = updater.new_scaled_da_gas_price;
+
+    // when
+    updater
+        .update_da_record_data(recorded_range, recorded_cost as u128)
+        .unwrap();
+
+    // then
+    let new_da_gas_price = updater.new_scaled_da_gas_price;
+    // because the profit is 10 and the da_p_component is 2, the new da gas price should be lesser than the previous one.
+    assert_eq!(new_da_gas_price, 0);
+    assert_ne!(old_da_gas_price, new_da_gas_price);
+}
+
+#[test]
+fn update_da_record_data__da_block_increases_da_gas_price() {
+    // given
+    let da_cost_per_byte = 40;
+    let da_recorded_block_height = 10;
+    let l2_block_height = 11;
+    let original_known_total_cost = 150;
+    let unrecorded_blocks = vec![BlockBytes {
+        height: 11,
+        block_bytes: 3000,
+    }];
+    let da_p_component = 2;
+    let guessed_cost: u64 = unrecorded_blocks
+        .iter()
+        .map(|block| block.block_bytes * da_cost_per_byte)
+        .sum();
+    let projected_total_cost = original_known_total_cost + guessed_cost;
+
+    let mut updater = UpdaterBuilder::new()
+        .with_da_cost_per_byte(da_cost_per_byte as u128)
+        .with_da_p_component(da_p_component)
+        .with_last_profit(-10, 0)
+        .with_da_recorded_block_height(da_recorded_block_height)
+        .with_l2_block_height(l2_block_height)
+        .with_projected_total_cost(projected_total_cost as u128)
+        .with_known_total_cost(original_known_total_cost as u128)
+        .with_unrecorded_blocks(unrecorded_blocks.clone())
+        .build();
+
+    let new_cost_per_byte = 100;
+    let (recorded_heights, recorded_cost) =
+        unrecorded_blocks
+            .iter()
+            .fold((vec![], 0), |(mut range, cost), block| {
+                range.push(block.height);
+                (range, cost + block.block_bytes * new_cost_per_byte)
+            });
+    let min = recorded_heights.iter().min().unwrap();
+    let max = recorded_heights.iter().max().unwrap();
+    let recorded_range = *min..(max + 1);
+
+    let old_da_gas_price = updater.new_scaled_da_gas_price;
+
+    // when
+    updater
+        .update_da_record_data(recorded_range, recorded_cost as u128)
+        .unwrap();
+
+    // then
+    let new_da_gas_price = updater.new_scaled_da_gas_price;
+    // because the profit is -10 and the da_p_component is 2, the new da gas price should be greater than the previous one.
+    assert_eq!(new_da_gas_price, 6);
+    assert_ne!(old_da_gas_price, new_da_gas_price);
+}
+
+#[test]
+fn update_da_record_data__da_block_will_not_change_da_gas_price() {
+    // given
+    let da_cost_per_byte = 40;
+    let da_recorded_block_height = 10;
+    let l2_block_height = 11;
+    let original_known_total_cost = 150;
+    let unrecorded_blocks = vec![BlockBytes {
+        height: 11,
+        block_bytes: 3000,
+    }];
+    let da_p_component = 2;
+    let guessed_cost: u64 = unrecorded_blocks
+        .iter()
+        .map(|block| block.block_bytes * da_cost_per_byte)
+        .sum();
+    let projected_total_cost = original_known_total_cost + guessed_cost;
+
+    let mut updater = UpdaterBuilder::new()
+        .with_da_cost_per_byte(da_cost_per_byte as u128)
+        .with_da_p_component(da_p_component)
+        .with_last_profit(0, 0)
+        .with_da_recorded_block_height(da_recorded_block_height)
+        .with_l2_block_height(l2_block_height)
+        .with_projected_total_cost(projected_total_cost as u128)
+        .with_known_total_cost(original_known_total_cost as u128)
+        .with_unrecorded_blocks(unrecorded_blocks.clone())
+        .build();
+
+    let new_cost_per_byte = 100;
+    let (recorded_heights, recorded_cost) =
+        unrecorded_blocks
+            .iter()
+            .fold((vec![], 0), |(mut range, cost), block| {
+                range.push(block.height);
+                (range, cost + block.block_bytes * new_cost_per_byte)
+            });
+    let min = recorded_heights.iter().min().unwrap();
+    let max = recorded_heights.iter().max().unwrap();
+    let recorded_range = *min..(max + 1);
+
+    let old_da_gas_price = updater.new_scaled_da_gas_price;
+
+    // when
+    updater
+        .update_da_record_data(recorded_range, recorded_cost as u128)
+        .unwrap();
+
+    // then
+    let new_da_gas_price = updater.new_scaled_da_gas_price;
+    assert_eq!(old_da_gas_price, new_da_gas_price);
+}

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

@@ -17,6 +17,10 @@ use crate::{
     ports::MetadataStorage,
 };
 use fuel_core_storage::{
+    codec::{
+        postcard::Postcard,
+        Encode,
+    },
     kv_store::KeyValueInspect,
     structured_storage::StructuredStorage,
     transactional::{
@@ -101,6 +105,8 @@ pub fn get_block_info(
         height: (*block.header().height()).into(),
         gas_used: used_gas,
         block_gas_capacity: block_gas_limit,
+        block_bytes: Postcard::encode(block).len() as u64,
+        block_fees: fee,
     };
     Ok(info)
 }

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

@@ -36,5 +36,9 @@ pub enum BlockInfo {
         gas_used: u64,
         // Total gas capacity of the block
         block_gas_capacity: u64,
+        // The size of block in bytes
+        block_bytes: u64,
+        // The fees the block has collected
+        block_fees: u64,
     },
 }

crates/services/gas_price_service/src/v0/service.rs

@@ -104,6 +104,7 @@ where
                 height,
                 gas_used,
                 block_gas_capacity,
+                ..
             } => {
                 self.handle_normal_block(height, gas_used, block_gas_capacity)
                     .await?;
@@ -225,6 +226,8 @@ mod tests {
             height: block_height,
             gas_used: 60,
             block_gas_capacity: 100,
+            block_bytes: 100,
+            block_fees: 100,
         };
 
         let (l2_block_sender, l2_block_receiver) = mpsc::channel(1);

crates/services/gas_price_service/src/v0/tests.rs

@@ -147,6 +147,8 @@ async fn next_gas_price__affected_by_new_l2_block() {
         height: 1,
         gas_used: 60,
         block_gas_capacity: 100,
+        block_bytes: 100,
+        block_fees: 100,
     };
     let (l2_block_sender, l2_block_receiver) = tokio::sync::mpsc::channel(1);
     let l2_block_source = FakeL2BlockSource {
@@ -186,6 +188,8 @@ async fn next__new_l2_block_saves_old_metadata() {
         height: 1,
         gas_used: 60,
         block_gas_capacity: 100,
+        block_bytes: 100,
+        block_fees: 100,
     };
     let (l2_block_sender, l2_block_receiver) = tokio::sync::mpsc::channel(1);
     let l2_block_source = FakeL2BlockSource {

crates/services/gas_price_service/src/v1.rs

@@ -1,3 +1,4 @@
 pub mod algorithm;
 pub mod da_source_service;
 pub mod metadata;
+pub mod service;

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

@@ -1,4 +1,7 @@
-use crate::common::gas_price_algorithm::GasPriceAlgorithm;
+use crate::common::gas_price_algorithm::{
+    GasPriceAlgorithm,
+    SharedGasPriceAlgo,
+};
 use fuel_core_types::fuel_types::BlockHeight;
 use fuel_gas_price_algorithm::v1::AlgorithmV1;
 
@@ -11,3 +14,5 @@ impl GasPriceAlgorithm for AlgorithmV1 {
         self.worst_case(block_height.into())
     }
 }
+
+pub type SharedV1Algorithm = SharedGasPriceAlgo<AlgorithmV1>;

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

@@ -7,7 +7,7 @@ pub mod service;
 
 #[derive(Debug, Default, Clone, Eq, Hash, PartialEq)]
 pub struct DaBlockCosts {
-    pub l2_block_range: core::ops::Range<u64>,
+    pub l2_block_range: core::ops::Range<u32>,
     pub blob_size_bytes: u32,
     pub blob_cost_wei: u128,
 }

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

@@ -21,12 +21,12 @@ trait BlockCommitterApi: Send + Sync {
     /// Used to get the costs for a specific seqno
     async fn get_costs_by_seqno(
         &self,
-        number: u64,
+        number: u32,
     ) -> DaBlockCostsResult<Option<RawDaBlockCosts>>;
     /// Used to get the costs for a range of blocks (inclusive)
     async fn get_cost_bundles_by_range(
         &self,
-        range: core::ops::Range<u64>,
+        range: core::ops::Range<u32>,
     ) -> DaBlockCostsResult<Vec<Option<RawDaBlockCosts>>>;
 }
 
@@ -40,9 +40,9 @@ pub struct BlockCommitterDaBlockCosts<BlockCommitter> {
 #[derive(Debug, Deserialize, Serialize, Clone, Default, PartialEq)]
 pub struct RawDaBlockCosts {
     /// Sequence number (Monotonically increasing nonce)
-    pub sequence_number: u64,
+    pub sequence_number: u32,
     /// The range of blocks that the costs apply to
-    pub blocks_range: core::ops::Range<u64>,
+    pub blocks_range: core::ops::Range<u32>,
     /// The DA block height of the last transaction for the range of blocks
     pub da_block_height: DaBlockHeight,
     /// Rolling sum cost of posting blobs (wei)
@@ -143,7 +143,7 @@ impl BlockCommitterApi for BlockCommitterHttpApi {
 
     async fn get_costs_by_seqno(
         &self,
-        number: u64,
+        number: u32,
     ) -> DaBlockCostsResult<Option<RawDaBlockCosts>> {
         let response = self
             .client
@@ -157,7 +157,7 @@ impl BlockCommitterApi for BlockCommitterHttpApi {
 
     async fn get_cost_bundles_by_range(
         &self,
-        range: core::ops::Range<u64>,
+        range: core::ops::Range<u32>,
     ) -> DaBlockCostsResult<Vec<Option<RawDaBlockCosts>>> {
         let response = self
             .client
@@ -192,23 +192,24 @@ mod tests {
         }
         async fn get_costs_by_seqno(
             &self,
-            seq_no: u64,
+            seq_no: u32,
         ) -> DaBlockCostsResult<Option<RawDaBlockCosts>> {
             // arbitrary logic to generate a new value
             let mut value = self.value.clone();
             if let Some(value) = &mut value {
                 value.sequence_number = seq_no;
                 value.blocks_range =
                     value.blocks_range.end * seq_no..value.blocks_range.end * seq_no + 10;
-                value.da_block_height = value.da_block_height + (seq_no + 1).into();
+                value.da_block_height =
+                    value.da_block_height + ((seq_no + 1) as u64).into();
                 value.total_cost += 1;
                 value.total_size_bytes += 1;
             }
             Ok(value)
         }
         async fn get_cost_bundles_by_range(
             &self,
-            _: core::ops::Range<u64>,
+            _: core::ops::Range<u32>,
         ) -> DaBlockCostsResult<Vec<Option<RawDaBlockCosts>>> {
             Ok(vec![self.value.clone()])
         }
@@ -286,7 +287,7 @@ mod tests {
         }
         async fn get_costs_by_seqno(
             &self,
-            seq_no: u64,
+            seq_no: u32,
         ) -> DaBlockCostsResult<Option<RawDaBlockCosts>> {
             // arbitrary logic to generate a new value
             let mut value = self.value.clone();
@@ -301,7 +302,7 @@ mod tests {
         }
         async fn get_cost_bundles_by_range(
             &self,
-            _: core::ops::Range<u64>,
+            _: core::ops::Range<u32>,
         ) -> DaBlockCostsResult<Vec<Option<RawDaBlockCosts>>> {
             Ok(vec![self.value.clone()])
         }

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

@@ -15,6 +15,7 @@ use tokio::{
 
 use crate::v1::da_source_service::DaBlockCosts;
 pub use anyhow::Result;
+use fuel_core_services::stream::BoxFuture;
 
 #[derive(Clone)]
 pub struct SharedState(Sender<DaBlockCosts>);