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/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>);
@@ -103,12 +104,14 @@ where
     async fn run(&mut self, state_watcher: &mut StateWatcher) -> Result<bool> {
         let continue_running;
 
+        let tick: BoxFuture<tokio::time::Instant> = Box::pin(self.poll_interval.tick());
+
         tokio::select! {
             biased;
             _ = state_watcher.while_started() => {
                 continue_running = false;
             }
-            _ = self.poll_interval.tick() => {
+            _ = tick => {
                 let da_block_costs = self.source.request_da_block_cost().await?;
                 self.shared_state.0.send(da_block_costs)?;
                 continue_running = true;

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

@@ -1,7 +1,6 @@
 use crate::v0::metadata::V0Metadata;
 use fuel_gas_price_algorithm::v1::{
     AlgorithmUpdaterV1,
-    ClampedPercentage,
     L2ActivityTracker,
 };
 use std::num::NonZeroU64;
@@ -41,7 +40,7 @@ pub struct V1Metadata {
 impl V1Metadata {
     pub fn construct_from_v0_metadata(
         v0_metadata: V0Metadata,
-        config: V1AlgorithmConfig,
+        config: &V1AlgorithmConfig,
     ) -> anyhow::Result<Self> {
         let metadata = Self {
             new_scaled_exec_price: v0_metadata
@@ -68,19 +67,56 @@ impl V1Metadata {
 }
 
 pub struct V1AlgorithmConfig {
-    new_exec_gas_price: u64,
-    min_exec_gas_price: u64,
-    exec_gas_price_change_percent: u16,
-    l2_block_fullness_threshold_percent: u8,
-    gas_price_factor: NonZeroU64,
-    min_da_gas_price: u64,
-    max_da_gas_price_change_percent: u16,
-    da_p_component: i64,
-    da_d_component: i64,
-    normal_range_size: u16,
-    capped_range_size: u16,
-    decrease_range_size: u16,
-    block_activity_threshold: u8,
+    pub new_exec_gas_price: u64,
+    pub min_exec_gas_price: u64,
+    pub exec_gas_price_change_percent: u16,
+    pub l2_block_fullness_threshold_percent: u8,
+    pub gas_price_factor: NonZeroU64,
+    pub min_da_gas_price: u64,
+    pub max_da_gas_price_change_percent: u16,
+    pub da_p_component: i64,
+    pub da_d_component: i64,
+    pub normal_range_size: u16,
+    pub capped_range_size: u16,
+    pub decrease_range_size: u16,
+    pub block_activity_threshold: u8,
+    pub unrecorded_blocks: Vec<(u32, u64)>,
+}
+
+impl From<&V1AlgorithmConfig> for AlgorithmUpdaterV1 {
+    fn from(value: &V1AlgorithmConfig) -> Self {
+        let l2_activity = L2ActivityTracker::new_full(
+            value.normal_range_size,
+            value.capped_range_size,
+            value.decrease_range_size,
+            value.block_activity_threshold.into(),
+        );
+        let unrecorded_blocks = value.unrecorded_blocks.clone().into_iter().collect();
+        Self {
+            new_scaled_exec_price: value.new_exec_gas_price,
+            l2_block_height: 0,
+            new_scaled_da_gas_price: value.min_da_gas_price,
+            gas_price_factor: value.gas_price_factor,
+            total_da_rewards_excess: 0,
+            da_recorded_block_height: 0,
+            latest_known_total_da_cost_excess: 0,
+            projected_total_da_cost: 0,
+            last_profit: 0,
+            second_to_last_profit: 0,
+            latest_da_cost_per_byte: 0,
+            l2_activity,
+            min_exec_gas_price: value.min_exec_gas_price,
+            exec_gas_price_change_percent: value.exec_gas_price_change_percent,
+            l2_block_fullness_threshold_percent: value
+                .l2_block_fullness_threshold_percent
+                .into(),
+            min_da_gas_price: value.min_da_gas_price,
+            max_da_gas_price_change_percent: value.max_da_gas_price_change_percent,
+            da_p_component: value.da_p_component,
+            da_d_component: value.da_d_component,
+            unrecorded_blocks,
+        }
+    }
 }
 
 impl From<AlgorithmUpdaterV1> for V1Metadata {
@@ -104,15 +140,19 @@ impl From<AlgorithmUpdaterV1> for V1Metadata {
 
 pub fn v1_algorithm_from_metadata(
     metadata: V1Metadata,
-    config: V1AlgorithmConfig,
+    config: &V1AlgorithmConfig,
 ) -> AlgorithmUpdaterV1 {
     let l2_activity = L2ActivityTracker::new_full(
         config.normal_range_size,
         config.capped_range_size,
         config.decrease_range_size,
         config.block_activity_threshold.into(),
     );
-    let unrecorded_blocks = metadata.unrecorded_blocks.into_iter().collect();
+    let unrecorded_blocks = metadata
+        .unrecorded_blocks
+        .into_iter()
+        .chain(config.unrecorded_blocks.clone())
+        .collect();
     AlgorithmUpdaterV1 {
         new_scaled_exec_price: metadata.new_scaled_exec_price,
         l2_block_height: metadata.l2_block_height,