diff --git a/Cargo.lock b/Cargo.lock index 7eb0dd52f43..3fedf8bdb41 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1609,12 +1609,12 @@ checksum = "5b63caa9aa9397e2d9480a9b13673856c78d8ac123288526c37d7839f2a86990" [[package]] name = "colored" -version = "2.1.0" +version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cbf2150cce219b664a8a70df7a1f933836724b503f8a413af9365b4dcc4d90b8" +checksum = "117725a109d387c937a1533ce01b450cbde6b88abceea8473c4d7a85853cda3c" dependencies = [ "lazy_static", - "windows-sys 0.48.0", + "windows-sys 0.59.0", ] [[package]] @@ -3465,10 +3465,12 @@ dependencies = [ "fuel-core-types 0.40.0", "fuel-gas-price-algorithm", "futures", + "mockito", "num_enum", "parking_lot", "reqwest 0.12.9", "serde", + "serde_json", "strum 0.25.0", "strum_macros 0.25.3", "thiserror 1.0.69", @@ -3745,7 +3747,6 @@ dependencies = [ "insta", "itertools 0.12.1", "k256", - "mockito", "postcard", "pretty_assertions", "primitive-types", diff --git a/crates/fuel-gas-price-algorithm/src/v1.rs b/crates/fuel-gas-price-algorithm/src/v1.rs index ad3718a89bb..7fe26ce10e3 100644 --- a/crates/fuel-gas-price-algorithm/src/v1.rs +++ b/crates/fuel-gas-price-algorithm/src/v1.rs @@ -3,7 +3,10 @@ use std::{ cmp::max, collections::BTreeMap, num::NonZeroU64, - ops::Div, + ops::{ + Div, + RangeInclusive, + }, }; #[cfg(test)] @@ -349,7 +352,7 @@ impl core::ops::Deref for ClampedPercentage { impl AlgorithmUpdaterV1 { pub fn update_da_record_data( &mut self, - heights: &[u32], + heights: RangeInclusive, recorded_bytes: u32, recording_cost: u128, unrecorded_blocks: &mut U, @@ -584,7 +587,7 @@ impl AlgorithmUpdaterV1 { fn da_block_update( &mut self, - heights: &[u32], + heights: RangeInclusive, recorded_bytes: u128, recording_cost: u128, unrecorded_blocks: &mut U, @@ -613,13 +616,13 @@ impl AlgorithmUpdaterV1 { // Always remove the blocks from the unrecorded blocks so they don't build up indefinitely fn update_unrecorded_block_bytes( &mut self, - heights: &[u32], + heights: RangeInclusive, unrecorded_blocks: &mut U, ) -> Result<(), Error> { let mut total: u128 = 0; for expected_height in heights { let maybe_bytes = unrecorded_blocks - .remove(expected_height) + .remove(&expected_height) .map_err(Error::CouldNotRemoveUnrecordedBlock)?; if let Some(bytes) = maybe_bytes { diff --git a/crates/fuel-gas-price-algorithm/src/v1/tests/update_da_record_data_tests.rs b/crates/fuel-gas-price-algorithm/src/v1/tests/update_da_record_data_tests.rs index 6d1feb220b4..da6812cce2f 100644 --- a/crates/fuel-gas-price-algorithm/src/v1/tests/update_da_record_data_tests.rs +++ b/crates/fuel-gas-price-algorithm/src/v1/tests/update_da_record_data_tests.rs @@ -5,7 +5,7 @@ use std::collections::BTreeMap; fn update_da_record_data__if_receives_batch_with_unknown_blocks_will_include_known_blocks_with_previous_cost( ) { // given - let recorded_heights: Vec = (1u32..3).collect(); + let recorded_heights = 1u32..=2; let recorded_cost = 1_000_000; let recorded_bytes = 500; let block_bytes = 1000; @@ -22,7 +22,7 @@ fn update_da_record_data__if_receives_batch_with_unknown_blocks_will_include_kno // when updater .update_da_record_data( - &recorded_heights, + recorded_heights, recorded_bytes, recorded_cost, &mut unrecorded_blocks, @@ -41,7 +41,7 @@ fn update_da_record_data__if_receives_batch_with_unknown_blocks_will_include_kno fn update_da_record_data__if_receives_batch_with_unknown_blocks_will_never_increase_cost_more_than_recorded_cost( ) { // given - let recorded_heights: Vec = (1u32..3).collect(); + let recorded_heights = 1u32..=2; let recorded_cost = 200; let block_bytes = 1000; let recorded_bytes = 500; @@ -58,7 +58,7 @@ fn update_da_record_data__if_receives_batch_with_unknown_blocks_will_never_incre // when updater .update_da_record_data( - &recorded_heights, + recorded_heights, recorded_bytes, recorded_cost, &mut unrecorded_blocks, @@ -87,11 +87,11 @@ fn update_da_record_data__updates_cost_per_byte() { let new_cost_per_byte = 100; let recorded_bytes = 500; let recorded_cost = (recorded_bytes * new_cost_per_byte) as u128; - let recorded_heights: Vec = (1u32..2).collect(); + let recorded_heights = 1u32..=1; // when updater .update_da_record_data( - &recorded_heights, + recorded_heights, recorded_bytes, recorded_cost, &mut unrecorded_blocks, @@ -121,13 +121,13 @@ fn update_da_record_data__updates_known_total_cost() { .with_unrecorded_blocks(&unrecorded_blocks) .build(); - let recorded_heights: Vec = (11u32..14).collect(); + let recorded_heights = 11u32..=13; let recorded_bytes = 500; let recorded_cost = 300; // when updater .update_da_record_data( - &recorded_heights, + recorded_heights, recorded_bytes, recorded_cost, &mut unrecorded_blocks, @@ -166,13 +166,13 @@ fn update_da_record_data__if_da_height_matches_l2_height_projected_and_known_mat let new_cost_per_byte = 100; let block_cost = block_bytes * new_cost_per_byte; - let recorded_heights: Vec = (11u32..14).collect(); + let recorded_heights = 11u32..=13; let recorded_bytes = 500; let recorded_cost = block_cost * 3; // when updater .update_da_record_data( - &recorded_heights, + recorded_heights, recorded_bytes, recorded_cost, &mut unrecorded_blocks, @@ -216,7 +216,7 @@ fn update_da_record_data__da_block_updates_projected_total_cost_with_known_and_g .build(); let new_cost_per_byte = 100; - let recorded_heights = vec![11, 12, 13]; + let recorded_heights = 11u32..=13; let recorded_bytes = 500; let recorded_cost = recorded_bytes * new_cost_per_byte; let recorded_bytes = 500; @@ -224,7 +224,7 @@ fn update_da_record_data__da_block_updates_projected_total_cost_with_known_and_g // when updater .update_da_record_data( - &recorded_heights, + recorded_heights, recorded_bytes, recorded_cost, &mut unrecorded_blocks, @@ -242,124 +242,6 @@ fn update_da_record_data__da_block_updates_projected_total_cost_with_known_and_g assert_eq!(actual, expected); } -#[test] -fn update_da_record_data__updates_known_total_cost_if_blocks_are_out_of_order() { - // given - let da_cost_per_byte = 20; - let block_bytes = 1000; - let mut unrecorded_blocks: BTreeMap<_, _> = - [(1, block_bytes), (2, block_bytes), (3, block_bytes)] - .into_iter() - .collect(); - let old_known_total_cost = 500; - let old_projected_total_cost = - old_known_total_cost + (block_bytes as u128 * da_cost_per_byte * 3); - let old_da_cost_per_byte = 20; - let mut updater = UpdaterBuilder::new() - .with_da_cost_per_byte(da_cost_per_byte) - .with_unrecorded_blocks(&unrecorded_blocks) - .with_da_cost_per_byte(old_da_cost_per_byte) - .with_known_total_cost(old_known_total_cost) - .with_projected_total_cost(old_projected_total_cost) - .build(); - let new_cost_per_byte = 100; - let recorded_bytes = 500; - let recorded_cost = recorded_bytes * new_cost_per_byte; - let recorded_heights: Vec = vec![3, 2]; - - // when - updater - .update_da_record_data( - &recorded_heights, - recorded_bytes, - recorded_cost as u128, - &mut unrecorded_blocks, - ) - .unwrap(); - - // then - let expected = updater.latest_known_total_da_cost_excess - + (block_bytes as u128 * new_cost_per_byte as u128); - let actual = updater.projected_total_da_cost; - assert_eq!(actual, expected); -} - -#[test] -fn update_da_record_data__updates_projected_total_cost_if_blocks_are_out_of_order() { - // given - let da_cost_per_byte = 20; - let block_bytes = 1000; - let mut unrecorded_blocks: BTreeMap<_, _> = - [(1, block_bytes), (2, block_bytes), (3, block_bytes)] - .into_iter() - .collect(); - let old_known_total_cost = 500; - let old_projected_total_cost = - old_known_total_cost + (block_bytes as u128 * da_cost_per_byte * 3); - let old_da_cost_per_byte = 20; - let mut updater = UpdaterBuilder::new() - .with_da_cost_per_byte(da_cost_per_byte) - .with_unrecorded_blocks(&unrecorded_blocks) - .with_da_cost_per_byte(old_da_cost_per_byte) - .with_known_total_cost(old_known_total_cost) - .with_projected_total_cost(old_projected_total_cost) - .build(); - let new_cost_per_byte = 100; - let recorded_bytes = 500; - let recorded_cost = recorded_bytes * new_cost_per_byte; - let recorded_heights: Vec = vec![3, 2]; - - // when - updater - .update_da_record_data( - &recorded_heights, - recorded_bytes, - recorded_cost as u128, - &mut unrecorded_blocks, - ) - .unwrap(); - - // then - let expected = updater.latest_known_total_da_cost_excess - + (block_bytes as u128 * new_cost_per_byte as u128); - let actual = updater.projected_total_da_cost; - assert_eq!(actual, expected); -} - -#[test] -fn update_da_record_data__updates_unrecorded_blocks() { - // given - let da_cost_per_byte = 20; - let block_bytes = 1000; - let mut unrecorded_blocks: BTreeMap<_, _> = - [(1, block_bytes), (2, block_bytes), (3, block_bytes)] - .into_iter() - .collect(); - let mut updater = UpdaterBuilder::new() - .with_da_cost_per_byte(da_cost_per_byte) - .with_unrecorded_blocks(&unrecorded_blocks) - .build(); - let new_cost_per_byte = 100; - let recorded_bytes = 500; - let recorded_cost = 2 * (recorded_bytes * new_cost_per_byte) as u128; - let recorded_heights: Vec = vec![3, 2]; - - // when - updater - .update_da_record_data( - &recorded_heights, - recorded_bytes, - recorded_cost, - &mut unrecorded_blocks, - ) - .unwrap(); - - // then - let expected = vec![(1, block_bytes)]; - let actual: Vec<_> = unrecorded_blocks.into_iter().collect(); - assert_eq!(actual, expected); -} - #[test] fn update_da_record_data__da_block_lowers_da_gas_price() { // given @@ -394,7 +276,7 @@ fn update_da_record_data__da_block_lowers_da_gas_price() { ); let min = *recorded_heights.iter().min().unwrap(); let max = *recorded_heights.iter().max().unwrap(); - let recorded_range: Vec = (*min..(max + 1)).collect(); + let recorded_range = *min..=*max; let recorded_bytes = 500; let old_da_gas_price = updater.new_scaled_da_gas_price; @@ -402,7 +284,7 @@ fn update_da_record_data__da_block_lowers_da_gas_price() { // when updater .update_da_record_data( - &recorded_range, + recorded_range, recorded_bytes, recorded_cost as u128, &mut unrecorded_blocks, @@ -451,7 +333,7 @@ fn update_da_record_data__da_block_increases_da_gas_price() { let min = *recorded_heights.iter().min().unwrap(); let max = *recorded_heights.iter().max().unwrap(); - let recorded_range: Vec = (*min..(max + 1)).collect(); + let recorded_range = *min..=*max; let recorded_bytes = 500; let old_da_gas_price = updater.new_scaled_da_gas_price; @@ -459,7 +341,7 @@ fn update_da_record_data__da_block_increases_da_gas_price() { // when updater .update_da_record_data( - &recorded_range, + recorded_range, recorded_bytes, recorded_cost as u128, &mut unrecorded_blocks, @@ -507,7 +389,7 @@ fn update_da_record_data__da_block_will_not_change_da_gas_price() { ); let min = *recorded_heights.iter().min().unwrap(); let max = *recorded_heights.iter().max().unwrap(); - let recorded_range: Vec = (*min..(max + 1)).collect(); + let recorded_range = *min..=*max; let recorded_bytes = 500; let old_da_gas_price = updater.new_scaled_da_gas_price; @@ -515,7 +397,7 @@ fn update_da_record_data__da_block_will_not_change_da_gas_price() { // when updater .update_da_record_data( - &recorded_range, + recorded_range, recorded_bytes, recorded_cost as u128, &mut unrecorded_blocks, diff --git a/crates/services/gas_price_service/Cargo.toml b/crates/services/gas_price_service/Cargo.toml index 4eab33d503b..efde85a485f 100644 --- a/crates/services/gas_price_service/Cargo.toml +++ b/crates/services/gas_price_service/Cargo.toml @@ -17,10 +17,12 @@ fuel-core-storage = { workspace = true, features = ["std"] } fuel-core-types = { workspace = true, features = ["std"] } fuel-gas-price-algorithm = { workspace = true } futures = { workspace = true } +mockito = { version = "1.6.1", optional = true } num_enum = { workspace = true } parking_lot = { workspace = true } reqwest = { workspace = true, features = ["json"] } serde = { workspace = true } +serde_json = { workspace = true, optional = true } strum = { workspace = true, features = ["derive"] } strum_macros = { workspace = true } thiserror = { workspace = true } @@ -32,4 +34,9 @@ tracing = { workspace = true } fuel-core-services = { workspace = true, features = ["test-helpers"] } fuel-core-storage = { workspace = true, features = ["test-helpers"] } fuel-core-types = { path = "./../../types", features = ["test-helpers"] } +mockito = { version = "1.6.1" } +serde_json = { workspace = true } tracing-subscriber = { workspace = true } + +[features] +test-helpers = ["dep:mockito", "dep:serde_json"] diff --git a/crates/services/gas_price_service/src/common/fuel_core_storage_adapter.rs b/crates/services/gas_price_service/src/common/fuel_core_storage_adapter.rs index 74bec1207d2..14b0675d762 100644 --- a/crates/services/gas_price_service/src/common/fuel_core_storage_adapter.rs +++ b/crates/services/gas_price_service/src/common/fuel_core_storage_adapter.rs @@ -1,9 +1,9 @@ use crate::{ common::{ fuel_core_storage_adapter::storage::{ - BundleIdTable, GasPriceColumn, GasPriceMetadata, + RecordedHeights, }, updater_metadata::UpdaterMetadata, utils::{ @@ -14,9 +14,9 @@ use crate::{ }, ports::{ GasPriceServiceAtomicStorage, - GetDaBundleId, + GetLatestRecordedHeight, GetMetadataStorage, - SetDaBundleId, + SetLatestRecordedHeight, SetMetadataStorage, }, }; @@ -99,25 +99,26 @@ where } } -impl GetDaBundleId for Storage +impl GetLatestRecordedHeight for Storage where Storage: Send + Sync, - Storage: StorageInspect, + Storage: StorageInspect, { - fn get_bundle_id(&self, block_height: &BlockHeight) -> GasPriceResult> { - let bundle_id = self - .storage::() - .get(block_height) + fn get_recorded_height(&self) -> GasPriceResult> { + const KEY: &() = &(); + let recorded_height = self + .storage::() + .get(KEY) .map_err(|err| GasPriceError::CouldNotFetchDARecord(err.into()))? .map(|no| *no); - Ok(bundle_id) + Ok(recorded_height) } } impl GasPriceServiceAtomicStorage for Storage where Storage: 'static, - Storage: GetMetadataStorage + GetDaBundleId, + Storage: GetMetadataStorage + GetLatestRecordedHeight, Storage: KeyValueInspect + Modifiable + Send + Sync, { type Transaction<'a> = StorageTransaction<&'a mut Storage> where Self: 'a; @@ -135,18 +136,18 @@ where } } -impl SetDaBundleId for Storage +impl SetLatestRecordedHeight for Storage where Storage: Send + Sync, - Storage: StorageMutate, + Storage: StorageMutate, { - fn set_bundle_id( + fn set_recorded_height( &mut self, - block_height: &BlockHeight, - bundle_id: u32, + recorded_height: BlockHeight, ) -> GasPriceResult<()> { - self.storage_as_mut::() - .insert(block_height, &bundle_id) + const KEY: &() = &(); + self.storage_as_mut::() + .insert(KEY, &recorded_height) .map_err(|err| GasPriceError::CouldNotFetchDARecord(err.into()))?; Ok(()) } diff --git a/crates/services/gas_price_service/src/common/fuel_core_storage_adapter/storage.rs b/crates/services/gas_price_service/src/common/fuel_core_storage_adapter/storage.rs index e9f20b411d8..61532696565 100644 --- a/crates/services/gas_price_service/src/common/fuel_core_storage_adapter/storage.rs +++ b/crates/services/gas_price_service/src/common/fuel_core_storage_adapter/storage.rs @@ -93,20 +93,18 @@ impl TableWithBlueprint for UnrecordedBlocksTable { } } -pub struct BundleIdTable; +/// Used to store the latest L2 block that has been recorded on the DA chain +pub struct RecordedHeights; -/// The sequence number or bundle id of the posted blocks. -type BundleId = u32; - -impl Mappable for BundleIdTable { +impl Mappable for RecordedHeights { type Key = Self::OwnedKey; - type OwnedKey = BlockHeight; + type OwnedKey = (); type Value = Self::OwnedValue; - type OwnedValue = BundleId; + type OwnedValue = BlockHeight; } -impl TableWithBlueprint for BundleIdTable { - type Blueprint = Plain, Postcard>; +impl TableWithBlueprint for RecordedHeights { + type Blueprint = Plain; type Column = GasPriceColumn; fn column() -> Self::Column { diff --git a/crates/services/gas_price_service/src/ports.rs b/crates/services/gas_price_service/src/ports.rs index 094f781f170..7670b3e0eab 100644 --- a/crates/services/gas_price_service/src/ports.rs +++ b/crates/services/gas_price_service/src/ports.rs @@ -35,26 +35,26 @@ pub trait GetMetadataStorage: Send + Sync { -> Result>; } -pub trait SetDaBundleId: Send + Sync { - fn set_bundle_id(&mut self, block_height: &BlockHeight, bundle_id: u32) - -> Result<()>; +pub trait SetLatestRecordedHeight: Send + Sync { + /// For any given L2 block produced, the DA will have committed some + fn set_recorded_height(&mut self, recorded_height: BlockHeight) -> Result<()>; } -pub trait GetDaBundleId: Send + Sync { - fn get_bundle_id(&self, block_height: &BlockHeight) -> Result>; +pub trait GetLatestRecordedHeight: Send + Sync { + fn get_recorded_height(&self) -> Result>; } pub trait GasPriceServiceAtomicStorage where Self: 'static, Self: Send + Sync, - Self: GetMetadataStorage + GetDaBundleId, + Self: GetMetadataStorage + GetLatestRecordedHeight, { type Transaction<'a>: AsUnrecordedBlocks + SetMetadataStorage + GetMetadataStorage - + SetDaBundleId - + GetDaBundleId + + SetLatestRecordedHeight + + GetLatestRecordedHeight where Self: 'a; diff --git a/crates/services/gas_price_service/src/v1/da_source_service.rs b/crates/services/gas_price_service/src/v1/da_source_service.rs index bae8d3d46bc..0c90b9a36d3 100644 --- a/crates/services/gas_price_service/src/v1/da_source_service.rs +++ b/crates/services/gas_price_service/src/v1/da_source_service.rs @@ -1,15 +1,18 @@ use crate::v1::da_source_service::service::DaBlockCostsSource; -use std::time::Duration; +use std::{ + ops::RangeInclusive, + time::Duration, +}; pub mod block_committer_costs; #[cfg(test)] pub mod dummy_costs; pub mod service; -#[derive(Debug, Default, Clone, Eq, Hash, PartialEq)] +#[derive(Debug, Clone, Eq, Hash, PartialEq)] pub struct DaBlockCosts { pub bundle_id: u32, - pub l2_blocks: Vec, + pub l2_blocks: RangeInclusive, pub bundle_size_bytes: u32, pub blob_cost_wei: u128, } @@ -33,7 +36,7 @@ mod tests { // given let expected_da_cost = DaBlockCosts { bundle_id: 1, - l2_blocks: (0..10).collect(), + l2_blocks: 0..=9, bundle_size_bytes: 1024 * 128, blob_cost_wei: 2, }; diff --git a/crates/services/gas_price_service/src/v1/da_source_service/block_committer_costs.rs b/crates/services/gas_price_service/src/v1/da_source_service/block_committer_costs.rs index ac888a0a70d..fb6077472c4 100644 --- a/crates/services/gas_price_service/src/v1/da_source_service/block_committer_costs.rs +++ b/crates/services/gas_price_service/src/v1/da_source_service/block_committer_costs.rs @@ -8,70 +8,84 @@ use crate::v1::da_source_service::{ DaBlockCosts, }; use anyhow::anyhow; -use fuel_core_types::blockchain::primitives::DaBlockHeight; +use fuel_core_types::{ + blockchain::primitives::DaBlockHeight, + fuel_types::BlockHeight, +}; use serde::{ Deserialize, Serialize, }; +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>; /// Used to get the costs for a specific seqno - async fn get_costs_by_seqno( - &self, - number: u32, - ) -> DaBlockCostsResult>; - /// Used to get the costs for a range of blocks (inclusive) - async fn get_cost_bundles_by_range( + async fn get_costs_by_l2_block_number( &self, - range: core::ops::Range, - ) -> DaBlockCostsResult>>; + l2_block_number: u32, + ) -> DaBlockCostsResult>; } /// This struct is used to denote the block committer da block costs source /// which receives data from the block committer (only http api for now) pub struct BlockCommitterDaBlockCosts { client: BlockCommitter, - last_raw_da_block_costs: Option, + last_recorded_height: Option, } #[derive(Debug, Deserialize, Serialize, Clone, Default, PartialEq)] pub struct RawDaBlockCosts { - /// Sequence number (Monotonically increasing nonce) - pub bundle_id: u32, - /// The range of blocks that the costs apply to - pub blocks_heights: Vec, + pub id: u32, + /// The beginning of the range of blocks that the costs apply to + pub start_height: u32, + /// The end of the range of blocks that the costs apply to + pub end_height: 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) - pub total_cost: u128, - /// Rolling sum size of blobs (bytes) - pub total_size_bytes: u32, + /// cost of posting this blob (wei) + pub cost: u128, + /// size of this blob (bytes) + pub size: u32, } impl From<&RawDaBlockCosts> for DaBlockCosts { fn from(raw_da_block_costs: &RawDaBlockCosts) -> Self { + let RawDaBlockCosts { + start_height, + end_height, + cost: cost_wei, + size: size_bytes, + id: bundle_id, + .. + } = *raw_da_block_costs; DaBlockCosts { - bundle_id: raw_da_block_costs.bundle_id, - l2_blocks: raw_da_block_costs - .blocks_heights - .clone() - .into_iter() - .collect(), - bundle_size_bytes: raw_da_block_costs.total_size_bytes, - blob_cost_wei: raw_da_block_costs.total_cost, + bundle_id, + // construct a vec of l2 blocks from the start_height to the end_height + l2_blocks: start_height..=end_height, + bundle_size_bytes: size_bytes, + blob_cost_wei: cost_wei, } } } +impl From for DaBlockCosts { + fn from(value: RawDaBlockCosts) -> Self { + Self::from(&value) + } +} + impl BlockCommitterDaBlockCosts { /// Create a new instance of the block committer da block costs source - pub fn new(client: BlockCommitter, last_value: Option) -> Self { + pub fn new( + client: BlockCommitter, + last_recorded_height: Option, + ) -> Self { Self { client, - last_raw_da_block_costs: last_value, + last_recorded_height, } } } @@ -81,49 +95,30 @@ impl DaBlockCostsSource for BlockCommitterDaBlockCosts DaBlockCostsResult> { - let raw_da_block_costs = match self.last_raw_da_block_costs { - Some(ref last_value) => { - self.client.get_costs_by_seqno(last_value.bundle_id + 1) - } - _ => self.client.get_latest_costs(), + async fn request_da_block_costs(&mut self) -> DaBlockCostsResult> { + let raw_da_block_costs: Vec<_> = + match self.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(), + }; + + tracing::info!("raw_da_block_costs: {:?}", raw_da_block_costs); + let da_block_costs: Vec<_> = + raw_da_block_costs.iter().map(DaBlockCosts::from).collect(); + tracing::info!("da_block_costs: {:?}", da_block_costs); + if let Some(cost) = raw_da_block_costs.last() { + self.last_recorded_height = Some(BlockHeight::from(cost.end_height)); } - .await?; - - let Some(ref raw_da_block_costs) = raw_da_block_costs else { - // TODO: This is really annoying if there haven't been any costs yet. Do we need this? - // Gonna return `Option::None` for now - // return Err(anyhow!("No response from block committer")) - return Ok(None) - }; - - let da_block_costs = self.last_raw_da_block_costs.iter().fold( - Ok(raw_da_block_costs.into()), - |costs: DaBlockCostsResult, last_value| { - let costs = costs.expect("Defined to be OK"); - let blob_size_bytes = costs - .bundle_size_bytes - .checked_sub(last_value.total_size_bytes) - .ok_or(anyhow!("Blob size bytes underflow"))?; - let blob_cost_wei = raw_da_block_costs - .total_cost - .checked_sub(last_value.total_cost) - .ok_or(anyhow!("Blob cost wei underflow"))?; - Ok(DaBlockCosts { - bundle_size_bytes: blob_size_bytes, - blob_cost_wei, - ..costs - }) - }, - )?; - self.last_raw_da_block_costs = Some(raw_da_block_costs.clone()); - Ok(Some(da_block_costs)) + Ok(da_block_costs) } - async fn set_last_value(&mut self, bundle_id: u32) -> DaBlockCostsResult<()> { - self.last_raw_da_block_costs = self.client.get_costs_by_seqno(bundle_id).await?; + + async fn set_last_value(&mut self, height: BlockHeight) -> DaBlockCostsResult<()> { + self.last_recorded_height = Some(height); Ok(()) } } @@ -142,54 +137,319 @@ impl BlockCommitterHttpApi { } } +const PAGE_SIZE: u32 = 10; #[async_trait::async_trait] impl BlockCommitterApi for BlockCommitterHttpApi { - async fn get_latest_costs(&self) -> DaBlockCostsResult> { + async fn get_costs_by_l2_block_number( + &self, + l2_block_number: u32, + ) -> DaBlockCostsResult> { + // Specific: http://localhost:8080/v1/costs?variant=specific&value=19098935&limit=5 if let Some(url) = &self.url { - let val = self.client.get(url).send().await?; - tracing::warn!("val: {:?}", val); - let response = val.json::>().await?; - tracing::warn!("Response: {:?}", response); - Ok(response) + tracing::info!("getting costs by l2 block number"); + let formatted_url = format!("{url}/v1/costs?variant=specific&value={l2_block_number}&limit={PAGE_SIZE}"); + tracing::info!("Formatted URL: {:?}", formatted_url); + let response = self.client.get(formatted_url).send().await?; + tracing::info!("response: {:?}", response); + let parsed = response.json::>().await?; + tracing::info!("parse: {:?}", parsed); + Ok(parsed) } else { - Ok(None) + Ok(vec![]) } } - async fn get_costs_by_seqno( - &self, - number: u32, - ) -> DaBlockCostsResult> { + async fn get_latest_costs(&self) -> DaBlockCostsResult> { + // Latest: http://localhost:8080/v1/costs?variant=latest&limit=5 if let Some(url) = &self.url { - let val = self - .client - .get(format!("{}/{}", url, number)) - .send() - .await?; - tracing::warn!("val: {:?}", val); - let response = val.json::>().await?; - tracing::warn!("Response: {:?}", response); - Ok(response) + tracing::info!("getting latest costs"); + let formatted_url = format!("{url}/v1/costs?variant=latest&limit=1"); + tracing::info!("Formatted URL: {:?}", formatted_url); + let response = self.client.get(formatted_url).send().await?; + tracing::info!("response: {:?}", response); + let raw_da_block_costs = response.json::>().await?; + tracing::info!("Parsed: {:?}", raw_da_block_costs); + // only take the first element, since we are only looking for the most recent + Ok(raw_da_block_costs.first().cloned()) } else { Ok(None) } } +} - async fn get_cost_bundles_by_range( - &self, - range: core::ops::Range, - ) -> DaBlockCostsResult>> { - if let Some(url) = &self.url { - let response = self - .client - .get(format!("{}/{}-{}", url, range.start, range.end)) - .send() - .await? - .json::>() - .await?; - Ok(response.into_iter().map(Some).collect()) - } else { - Ok(vec![]) +#[cfg(test)] +mod test_block_committer_http_api { + #![allow(non_snake_case)] + + use super::*; + use fake_server::FakeServer; + + #[test] + fn get_costs_by_l2_block_number__when_url_is_none__then_returns_empty_vec() { + let rt = tokio::runtime::Runtime::new().unwrap(); + + // given + let block_committer = BlockCommitterHttpApi::new(None); + let l2_block_number = 1; + + // when + let actual = rt.block_on(async { + block_committer + .get_costs_by_l2_block_number(l2_block_number) + .await + .unwrap() + }); + + // then + assert_eq!(actual.len(), 0); + } + #[test] + fn get_costs_by_l2_block_number__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 l2_block_number = 51; + let block_committer = BlockCommitterHttpApi::new(Some(url)); + + let too_early_count = 5; + let do_not_fit_count = 5; + + let mut current_height = 0; + let mut bundle_id = 0; + let mut da_block_height: u64 = 0; + + // shouldn't return + for _ in 0..too_early_count { + bundle_id += 1; + da_block_height += 1; + current_height += 1; + let start_height = current_height; + current_height += 9; + let end_height = current_height; + let costs = RawDaBlockCosts { + id: bundle_id, + start_height, + end_height, + da_block_height: DaBlockHeight::from(da_block_height), + cost: 1, + size: 1, + }; + mock.add_response(costs); + } + let mut expected = Vec::new(); + + // should return + for _ in 0..PAGE_SIZE { + bundle_id += 1; + da_block_height += 1; + current_height += 1; + let start_height = current_height; + current_height += 9; + let end_height = current_height; + let costs = RawDaBlockCosts { + id: bundle_id, + start_height, + end_height, + da_block_height: DaBlockHeight::from(da_block_height), + cost: 1, + size: 1, + }; + mock.add_response(costs.clone()); + expected.push(costs); + } + // don't fit + for _ in 0..do_not_fit_count { + bundle_id += 1; + da_block_height += 1; + current_height += 1; + let start_height = current_height; + current_height += 9; + let end_height = current_height; + let costs = RawDaBlockCosts { + id: bundle_id, + start_height, + end_height, + da_block_height: DaBlockHeight::from(da_block_height), + cost: 1, + size: 1, + }; + mock.add_response(costs); + } + + // when + mock.init(); + let actual = rt.block_on(async { + block_committer + .get_costs_by_l2_block_number(l2_block_number) + .await + .unwrap() + }); + + // 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 { + use crate::v1::da_source_service::block_committer_costs::RawDaBlockCosts; + use mockito::Matcher::Any; + use std::{ + collections::HashMap, + sync::{ + Arc, + Mutex, + }, + }; + + pub struct FakeServer { + server: mockito::ServerGuard, + responses: Arc>>, + } + + impl FakeServer { + pub fn new() -> Self { + let server = mockito::Server::new(); + let responses = Arc::new(Mutex::new(Vec::new())); + let mut fake = Self { server, responses }; + fake.init(); + fake + } + + #[allow(unused_variables)] + pub fn init(&mut self) { + let shared_responses = self.responses.clone(); + self.server + .mock("GET", Any) + .with_status(201) + .with_body_from_request(move |request| { + // take the requested number and return the corresponding response from the `responses` hashmap + let path = request.path_and_query(); + tracing::info!("Path: {:?}", path); + let query = path.split("variant=").last().unwrap(); + tracing::info!("Query: {:?}", query); + let mut values = query.split('&'); + let variant = values.next().unwrap(); + tracing::info!("Variant: {:?}", variant); + match variant { + // Latest: http://localhost:8080/v1/costs?variant=latest&limit=5 + // We don't support `limit` yet!!!! + "latest" => { + let args = values.next().unwrap(); + let limit = + args.split('=').last().unwrap().parse::().unwrap(); + assert!(limit == 1); + let guard = shared_responses.lock().unwrap(); + let most_recent = guard + .iter() + .fold(None, |acc, x| match acc { + None => Some(x), + Some(acc) => { + if x.end_height > acc.end_height { + Some(x) + } else { + Some(acc) + } + } + }) + .cloned(); + let response: Vec = + most_recent.into_iter().collect(); + serde_json::to_string(&response).unwrap().into() + } + // Specific: http://localhost:8080/v1/costs?variant=specific&value=19098935&limit=5 + "specific" => { + let args = values.next().unwrap(); + let mut specific_values = args.split('='); + let height = + specific_values.last().unwrap().parse::().unwrap(); + tracing::info!("Height: {:?}", height); + let maybe_limit = values + .next() + .and_then(|x| x.split('=').last()) + .and_then(|x| x.parse::().ok()); + tracing::info!("Limit: {:?}", maybe_limit); + let guard = shared_responses.lock().unwrap(); + let response = guard + .iter() + .filter(|costs| costs.end_height >= height) + .take(maybe_limit.unwrap_or(usize::MAX)) + .cloned() + .collect::>(); + serde_json::to_string(&response).unwrap().into() + } + _ => { + panic!("Invalid variant: {}", variant); + } + } + }) + .expect_at_least(1) + .create(); + } + + pub fn add_response(&mut self, costs: RawDaBlockCosts) { + let mut guard = self.responses.lock().unwrap(); + guard.push(costs); + } + + pub fn url(&self) -> String { + self.server.url() + } + } + impl Default for FakeServer { + fn default() -> Self { + Self::new() } } } @@ -214,41 +474,32 @@ mod tests { async fn get_latest_costs(&self) -> DaBlockCostsResult> { Ok(self.value.clone()) } - async fn get_costs_by_seqno( + async fn get_costs_by_l2_block_number( &self, - seq_no: u32, - ) -> DaBlockCostsResult> { + l2_block_number: u32, + ) -> DaBlockCostsResult> { // arbitrary logic to generate a new value let mut value = self.value.clone(); if let Some(value) = &mut value { - value.bundle_id = seq_no; - value.blocks_heights = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] - .to_vec() - .iter() - .map(|x| x * seq_no) - .collect(); + value.start_height = l2_block_number; + value.end_height = value.end_height + l2_block_number + 10; value.da_block_height = - value.da_block_height + ((seq_no + 1) as u64).into(); - value.total_cost += 1; - value.total_size_bytes += 1; + value.da_block_height + ((l2_block_number + 1) as u64).into(); + value.cost += 1; + value.size += 1; } - Ok(value) - } - async fn get_cost_bundles_by_range( - &self, - _: core::ops::Range, - ) -> DaBlockCostsResult>> { - Ok(vec![self.value.clone()]) + Ok(value.into_iter().collect()) } } fn test_da_block_costs() -> RawDaBlockCosts { RawDaBlockCosts { - bundle_id: 1, - blocks_heights: (0..10).collect(), + id: 1, + start_height: 1, + end_height: 10, da_block_height: 1u64.into(), - total_cost: 1, - total_size_bytes: 1, + cost: 1, + size: 1, } } @@ -257,98 +508,34 @@ mod tests { ) { // given let da_block_costs = test_da_block_costs(); - let expected = Some((&da_block_costs).into()); + let expected = vec![(&da_block_costs).into()]; let mock_api = MockBlockCommitterApi::new(Some(da_block_costs)); let mut block_committer = BlockCommitterDaBlockCosts::new(mock_api, None); // when - let actual = block_committer.request_da_block_cost().await.unwrap(); + let actual = block_committer.request_da_block_costs().await.unwrap(); // then assert_eq!(actual, expected); } #[tokio::test] - async fn request_da_block_cost__when_last_value_is_some__then_get_costs_by_seqno_is_called( + async fn request_da_block_cost__when_last_value_is_some__then_get_costs_by_l2_block_number_is_called( ) { // given let mut da_block_costs = test_da_block_costs(); + let da_block_costs_len = da_block_costs.end_height - da_block_costs.start_height; let mock_api = MockBlockCommitterApi::new(Some(da_block_costs.clone())); + let latest_height = BlockHeight::new(da_block_costs.end_height); let mut block_committer = - BlockCommitterDaBlockCosts::new(mock_api, Some(da_block_costs.clone())); - - // when - let actual = block_committer.request_da_block_cost().await.unwrap(); - - // then - assert_ne!(da_block_costs.blocks_heights, actual.unwrap().l2_blocks); - } - - // TODO: Do we need this? - // #[tokio::test] - // async fn request_da_block_cost__when_response_is_none__then_error() { - // // given - // let mock_api = MockBlockCommitterApi::new(None); - // let mut block_committer = BlockCommitterDaBlockCosts::new(mock_api, None); - // - // // when - // let result = block_committer.request_da_block_cost().await; - // - // // then - // assert!(result.is_err()); - // } - - struct UnderflowingMockBlockCommitterApi { - value: Option, - } - - impl UnderflowingMockBlockCommitterApi { - fn new(value: Option) -> Self { - Self { value } - } - } - - #[async_trait::async_trait] - impl BlockCommitterApi for UnderflowingMockBlockCommitterApi { - async fn get_latest_costs(&self) -> DaBlockCostsResult> { - Ok(self.value.clone()) - } - async fn get_costs_by_seqno( - &self, - seq_no: u32, - ) -> DaBlockCostsResult> { - // arbitrary logic to generate a new value - let mut value = self.value.clone(); - if let Some(value) = &mut value { - value.bundle_id = seq_no; - value.blocks_heights = - value.blocks_heights.iter().map(|x| x + seq_no).collect(); - value.da_block_height = value.da_block_height + 1u64.into(); - value.total_cost -= 1; - value.total_size_bytes -= 1; - } - Ok(value) - } - async fn get_cost_bundles_by_range( - &self, - _: core::ops::Range, - ) -> DaBlockCostsResult>> { - Ok(vec![self.value.clone()]) - } - } - - #[tokio::test] - async fn request_da_block_cost__when_underflow__then_error() { - // given - let da_block_costs = test_da_block_costs(); - let mock_api = UnderflowingMockBlockCommitterApi::new(Some(da_block_costs)); - let mut block_committer = BlockCommitterDaBlockCosts::new(mock_api, None); - let _ = block_committer.request_da_block_cost().await.unwrap(); + BlockCommitterDaBlockCosts::new(mock_api, Some(latest_height)); // when - let result = block_committer.request_da_block_cost().await; + let actual = block_committer.request_da_block_costs().await.unwrap(); // then - assert!(result.is_err()); + let l2_blocks = actual.first().unwrap().l2_blocks.clone(); + let range_len = l2_blocks.end() - l2_blocks.start(); + assert_ne!(da_block_costs_len, range_len); } } diff --git a/crates/services/gas_price_service/src/v1/da_source_service/dummy_costs.rs b/crates/services/gas_price_service/src/v1/da_source_service/dummy_costs.rs index 93c02f583f4..8b5094d12e4 100644 --- a/crates/services/gas_price_service/src/v1/da_source_service/dummy_costs.rs +++ b/crates/services/gas_price_service/src/v1/da_source_service/dummy_costs.rs @@ -5,6 +5,7 @@ use crate::v1::da_source_service::{ }, DaBlockCosts, }; +use fuel_core_types::fuel_types::BlockHeight; use std::sync::Arc; use tokio::sync::Notify; @@ -22,13 +23,11 @@ impl DummyDaBlockCosts { #[async_trait::async_trait] impl DaBlockCostsSource for DummyDaBlockCosts { - async fn request_da_block_cost( - &mut self, - ) -> DaBlockCostsResult> { + async fn request_da_block_costs(&mut self) -> DaBlockCostsResult> { match &self.value { Ok(da_block_costs) => { self.notifier.notify_waiters(); - Ok(Some(da_block_costs.clone())) + Ok(vec![da_block_costs.clone()]) } Err(err) => { self.notifier.notify_waiters(); @@ -37,7 +36,7 @@ impl DaBlockCostsSource for DummyDaBlockCosts { } } - async fn set_last_value(&mut self, _bundle_id: u32) -> DaBlockCostsResult<()> { + async fn set_last_value(&mut self, _height: BlockHeight) -> DaBlockCostsResult<()> { unimplemented!("This is a dummy implementation"); } } diff --git a/crates/services/gas_price_service/src/v1/da_source_service/service.rs b/crates/services/gas_price_service/src/v1/da_source_service/service.rs index 0c36c3012fc..675d82cda55 100644 --- a/crates/services/gas_price_service/src/v1/da_source_service/service.rs +++ b/crates/services/gas_price_service/src/v1/da_source_service/service.rs @@ -16,6 +16,7 @@ use tokio::{ use crate::v1::da_source_service::DaBlockCosts; pub use anyhow::Result; +use fuel_core_types::fuel_types::BlockHeight; #[derive(Clone)] pub struct SharedState(Sender); @@ -58,9 +59,10 @@ where } async fn process_block_costs(&mut self) -> Result<()> { - let da_block_costs_res = self.source.request_da_block_cost().await; + let da_block_costs_res = self.source.request_da_block_costs().await; tracing::debug!("Received block costs: {:?}", da_block_costs_res); - if let Some(da_block_costs) = da_block_costs_res? { + let da_block_costs = da_block_costs_res?; + for da_block_costs in da_block_costs { self.shared_state.0.send(da_block_costs)?; } Ok(()) @@ -71,8 +73,8 @@ where /// da block costs in a way they see fit #[async_trait::async_trait] pub trait DaBlockCostsSource: Send + Sync { - async fn request_da_block_cost(&mut self) -> Result>; - async fn set_last_value(&mut self, bundle_id: u32) -> Result<()>; + async fn request_da_block_costs(&mut self) -> Result>; + async fn set_last_value(&mut self, block_height: BlockHeight) -> Result<()>; } #[async_trait::async_trait] diff --git a/crates/services/gas_price_service/src/v1/service.rs b/crates/services/gas_price_service/src/v1/service.rs index 679d6372646..1c773c3a7e0 100644 --- a/crates/services/gas_price_service/src/v1/service.rs +++ b/crates/services/gas_price_service/src/v1/service.rs @@ -12,9 +12,9 @@ use crate::{ }, ports::{ GasPriceServiceAtomicStorage, - GetDaBundleId, + GetLatestRecordedHeight, GetMetadataStorage, - SetDaBundleId, + SetLatestRecordedHeight, SetMetadataStorage, }, v0::metadata::V0Metadata, @@ -161,31 +161,26 @@ where ) -> anyhow::Result<()> { let capacity = Self::validate_block_gas_capacity(block_gas_capacity)?; let mut storage_tx = self.storage_tx_provider.begin_transaction()?; - let prev_height = height.saturating_sub(1); - let mut bundle_id = storage_tx - .get_bundle_id(&BlockHeight::from(prev_height)) + let mut latest_recorded_height = storage_tx + .get_recorded_height() .map_err(|err| anyhow!(err))?; for da_block_costs in &self.da_block_costs_buffer { tracing::debug!("Updating DA block costs: {:?}", da_block_costs); + let l2_blocks = da_block_costs.l2_blocks.clone(); + let end = *l2_blocks.end(); self.algorithm_updater.update_da_record_data( - &da_block_costs.l2_blocks, + l2_blocks, da_block_costs.bundle_size_bytes, da_block_costs.blob_cost_wei, &mut storage_tx.as_unrecorded_blocks(), )?; - bundle_id = Some(da_block_costs.bundle_id); + latest_recorded_height = Some(BlockHeight::from(end)); } - if let Some(bundle_id) = bundle_id { + if let Some(recorded_height) = latest_recorded_height { storage_tx - .set_bundle_id(&BlockHeight::from(height), bundle_id) - .map_err(|err| anyhow!(err))?; - } - - if let Some(bundle_id) = bundle_id { - storage_tx - .set_bundle_id(&BlockHeight::from(height), bundle_id) + .set_recorded_height(recorded_height) .map_err(|err| anyhow!(err))?; } @@ -372,9 +367,9 @@ mod tests { use crate::{ common::{ fuel_core_storage_adapter::storage::{ - BundleIdTable, GasPriceColumn, GasPriceColumn::UnrecordedBlocks, + RecordedHeights, UnrecordedBlocksTable, }, gas_price_algorithm::SharedGasPriceAlgo, @@ -576,7 +571,7 @@ mod tests { DummyDaBlockCosts::new( Ok(DaBlockCosts { bundle_id: 1, - l2_blocks: (1..2).collect(), + l2_blocks: 1..=1, blob_cost_wei: u128::MAX, // Very expensive to trigger a change bundle_size_bytes: 3000, }), @@ -632,11 +627,11 @@ mod tests { } #[tokio::test] - async fn run__responses_from_da_service_update_bundle_id_in_storage() { + async fn run__responses_from_da_service_update_recorded_height_in_storage() { // given - let bundle_id = 1234; - let block_height = 2; - let l2_block_2 = BlockInfo::Block { + let recorded_block_height = 100; + let block_height = 200; + let l2_block = BlockInfo::Block { height: block_height, gas_used: 60, block_gas_capacity: 100, @@ -669,8 +664,8 @@ mod tests { let da_source = DaSourceService::new( DummyDaBlockCosts::new( Ok(DaBlockCosts { - bundle_id, - l2_blocks: (1..2).collect(), + bundle_id: 8765, + l2_blocks: 1..=recorded_block_height, blob_cost_wei: 9000, bundle_size_bytes: 3000, }), @@ -694,20 +689,23 @@ mod tests { service.run(&mut watcher).await; tokio::time::sleep(Duration::from_millis(100)).await; - l2_block_sender.send(l2_block_2).await.unwrap(); + l2_block_sender.send(l2_block).await.unwrap(); // when service.run(&mut watcher).await; tokio::time::sleep(Duration::from_millis(100)).await; // then - let latest_bundle_id = service + let latest_recorded_block_height = service .storage_tx_provider - .storage::() - .get(&BlockHeight::from(block_height)) + .storage::() + .get(&()) .unwrap() .unwrap(); - assert_eq!(*latest_bundle_id, bundle_id); + assert_eq!( + *latest_recorded_block_height, + BlockHeight::from(recorded_block_height) + ); service.shutdown().await.unwrap(); } diff --git a/crates/services/gas_price_service/src/v1/tests.rs b/crates/services/gas_price_service/src/v1/tests.rs index 349a090e97c..1a20b98fb5c 100644 --- a/crates/services/gas_price_service/src/v1/tests.rs +++ b/crates/services/gas_price_service/src/v1/tests.rs @@ -20,10 +20,10 @@ use crate::{ ports::{ GasPriceData, GasPriceServiceAtomicStorage, - GetDaBundleId, + GetLatestRecordedHeight, GetMetadataStorage, L2Data, - SetDaBundleId, + SetLatestRecordedHeight, SetMetadataStorage, }, v1::{ @@ -170,8 +170,8 @@ impl GetMetadataStorage for ErroringPersistedData { } } -impl GetDaBundleId for ErroringPersistedData { - fn get_bundle_id(&self, _block_height: &BlockHeight) -> GasPriceResult> { +impl GetLatestRecordedHeight for ErroringPersistedData { + fn get_recorded_height(&self) -> GasPriceResult> { Err(GasPriceError::CouldNotFetchDARecord(anyhow!("boo!"))) } } @@ -215,18 +215,14 @@ impl UnrecordedBlocks for UnimplementedStorageTx { } } -impl SetDaBundleId for UnimplementedStorageTx { - fn set_bundle_id( - &mut self, - _block_height: &BlockHeight, - _bundle_id: u32, - ) -> GasPriceResult<()> { +impl SetLatestRecordedHeight for UnimplementedStorageTx { + fn set_recorded_height(&mut self, _bundle_id: BlockHeight) -> GasPriceResult<()> { unimplemented!() } } -impl GetDaBundleId for UnimplementedStorageTx { - fn get_bundle_id(&self, _block_height: &BlockHeight) -> GasPriceResult> { +impl GetLatestRecordedHeight for UnimplementedStorageTx { + fn get_recorded_height(&self) -> GasPriceResult> { unimplemented!() } } @@ -243,7 +239,7 @@ impl AsUnrecordedBlocks for UnimplementedStorageTx { struct FakeDABlockCost { da_block_costs: Receiver, - bundle_id: Arc>>, + latest_received_height: Arc>>, } impl FakeDABlockCost { @@ -251,37 +247,38 @@ impl FakeDABlockCost { let (_sender, receiver) = tokio::sync::mpsc::channel(1); Self { da_block_costs: receiver, - bundle_id: Arc::new(Mutex::new(None)), + latest_received_height: Arc::new(Mutex::new(None)), } } fn new(da_block_costs: Receiver) -> Self { Self { da_block_costs, - bundle_id: Arc::new(Mutex::new(None)), + latest_received_height: Arc::new(Mutex::new(None)), } } - fn never_returns_with_handle_to_bundle_id() -> (Self, Arc>>) { + fn never_returns_with_handle_to_last_height( + ) -> (Self, Arc>>) { let (_sender, receiver) = tokio::sync::mpsc::channel(1); - let bundle_id = Arc::new(Mutex::new(None)); + let height = Arc::new(Mutex::new(None)); let service = Self { da_block_costs: receiver, - bundle_id: bundle_id.clone(), + latest_received_height: height.clone(), }; - (service, bundle_id) + (service, height) } } #[async_trait::async_trait] impl DaBlockCostsSource for FakeDABlockCost { - async fn request_da_block_cost(&mut self) -> Result> { + async fn request_da_block_costs(&mut self) -> Result> { let costs = self.da_block_costs.recv().await.unwrap(); - Ok(Some(costs)) + Ok(vec![costs]) } - async fn set_last_value(&mut self, bundle_id: u32) -> Result<()> { - self.bundle_id.lock().unwrap().replace(bundle_id); + async fn set_last_value(&mut self, height: BlockHeight) -> Result<()> { + self.latest_received_height.lock().unwrap().replace(height); Ok(()) } } @@ -696,8 +693,8 @@ async fn uninitialized_task__new__should_fail_if_cannot_fetch_metadata() { #[tokio::test] async fn uninitialized_task__init__starts_da_service_with_bundle_id_in_storage() { // given - let block_height = 1; - let bundle_id: u32 = 123; + let block_height = 100; + let recorded_height: u32 = 200; let original_metadata = arbitrary_metadata(); let different_config = different_arb_config(); @@ -708,11 +705,12 @@ async fn uninitialized_task__init__starts_da_service_with_bundle_id_in_storage() let settings = FakeSettings::default(); let block_stream = empty_block_stream(); let on_chain_db = FakeOnChainDb::new(different_l2_block); - let (da_cost_source, bundle_id_handle) = - FakeDABlockCost::never_returns_with_handle_to_bundle_id(); + let (da_cost_source, recorded_block_height_handle) = + FakeDABlockCost::never_returns_with_handle_to_last_height(); let mut inner = gas_price_database_with_metadata(&original_metadata); let mut tx = inner.begin_transaction().unwrap(); - tx.set_bundle_id(&block_height.into(), bundle_id).unwrap(); + tx.set_recorded_height(BlockHeight::from(recorded_height)) + .unwrap(); StorageTransaction::commit_transaction(tx).unwrap(); let service = UninitializedTask::new( different_config.clone(), @@ -730,8 +728,8 @@ async fn uninitialized_task__init__starts_da_service_with_bundle_id_in_storage() service.init(&StateWatcher::started()).await.unwrap(); // then - let actual = bundle_id_handle.lock().unwrap(); - let expected = Some(bundle_id); + let actual = recorded_block_height_handle.lock().unwrap(); + let expected = Some(BlockHeight::new(recorded_height)); assert_eq!(*actual, expected); } diff --git a/crates/services/gas_price_service/src/v1/uninitialized_task.rs b/crates/services/gas_price_service/src/v1/uninitialized_task.rs index 10ca224fdb4..db77274ccca 100644 --- a/crates/services/gas_price_service/src/v1/uninitialized_task.rs +++ b/crates/services/gas_price_service/src/v1/uninitialized_task.rs @@ -21,10 +21,10 @@ use crate::{ GasPriceData, GasPriceServiceAtomicStorage, GasPriceServiceConfig, - GetDaBundleId, + GetLatestRecordedHeight, GetMetadataStorage, L2Data, - SetDaBundleId, + SetLatestRecordedHeight, SetMetadataStorage, }, v1::{ @@ -157,21 +157,14 @@ where self.block_stream, ); - if let Some(bundle_id) = - self.gas_price_db.get_bundle_id(&self.gas_metadata_height)? - { - self.da_source.set_last_value(bundle_id).await?; + if let Some(last_recorded_height) = self.gas_price_db.get_recorded_height()? { + self.da_source.set_last_value(last_recorded_height).await?; + tracing::info!("Set last recorded height to {}", last_recorded_height); } let poll_duration = self .config .da_poll_interval .map(|x| Duration::from_millis(x.into())); - // TODO: Dupe code - if let Some(bundle_id) = - self.gas_price_db.get_bundle_id(&self.gas_metadata_height)? - { - self.da_source.set_last_value(bundle_id).await?; - } let da_service = DaSourceService::new(self.da_source, poll_duration); let da_service_runner = ServiceRunner::new(da_service); da_service_runner.start_and_await().await?; diff --git a/tests/Cargo.toml b/tests/Cargo.toml index bae9c54fc2d..6e9a03ce96d 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -38,7 +38,9 @@ fuel-core-bin = { path = "../bin/fuel-core", features = ["parquet", "p2p"] } fuel-core-client = { path = "../crates/client", features = ["test-helpers"] } fuel-core-compression = { path = "../crates/compression" } fuel-core-executor = { workspace = true, features = ["test-helpers"] } -fuel-core-gas-price-service = { path = "../crates/services/gas_price_service" } +fuel-core-gas-price-service = { path = "../crates/services/gas_price_service", features = [ + "test-helpers", +] } fuel-core-p2p = { path = "../crates/services/p2p", features = [ "test-helpers", ], optional = true } @@ -59,7 +61,6 @@ hyper = { workspace = true, features = ["server"] } insta = { workspace = true } itertools = { workspace = true } k256 = { version = "0.13.3", features = ["ecdsa-core"] } -mockito = "1.6.1" postcard = { workspace = true } primitive-types = { workspace = true, default-features = false } rand = { workspace = true } diff --git a/tests/tests/gas_price.rs b/tests/tests/gas_price.rs index 6ac0cf772cd..9d4ab9bff72 100644 --- a/tests/tests/gas_price.rs +++ b/tests/tests/gas_price.rs @@ -26,12 +26,19 @@ use fuel_core_client::client::{ FuelClient, }; use fuel_core_gas_price_service::{ - common::fuel_core_storage_adapter::storage::GasPriceMetadata, + common::fuel_core_storage_adapter::storage::{ + GasPriceMetadata, + RecordedHeights, + }, ports::{ GasPriceData, + GetLatestRecordedHeight, GetMetadataStorage, }, - v1::metadata::V1Metadata, + v1::{ + da_source_service::block_committer_costs::fake_server::FakeServer, + metadata::V1Metadata, + }, }; use fuel_core_poa::Trigger; use fuel_core_storage::{ @@ -54,7 +61,6 @@ use fuel_core_types::{ }, services::executor::TransactionExecutionResult, }; -use mockito::Matcher::Any; use rand::Rng; use std::{ collections::HashMap, @@ -285,24 +291,25 @@ async fn estimate_gas_price__is_greater_than_actual_price_at_desired_height() { assert!(estimated >= real); } -#[tokio::test] -async fn estimate_gas_price__returns_min_gas_price_if_starting_gas_price_is_zero() { - const MIN_GAS_PRICE: u64 = 1; - - // Given - let mut node_config = Config::local_node(); - node_config.min_exec_gas_price = MIN_GAS_PRICE; - node_config.starting_exec_gas_price = 0; - let srv = FuelService::new_node(node_config.clone()).await.unwrap(); - let client = FuelClient::from(srv.bound_address); - - // When - let result = client.estimate_gas_price(10).await.unwrap(); - - // Then - let actual = result.gas_price.0; - assert_eq!(MIN_GAS_PRICE, actual) -} +// TODO: this behavior is changing with https://github.com/FuelLabs/fuel-core/pull/2501 +// #[tokio::test] +// async fn estimate_gas_price__returns_min_gas_price_if_starting_gas_price_is_zero() { +// const MIN_GAS_PRICE: u64 = 1; +// +// // Given +// let mut node_config = Config::local_node(); +// node_config.min_exec_gas_price = MIN_GAS_PRICE; +// node_config.starting_exec_gas_price = 0; +// let srv = FuelService::new_node(node_config.clone()).await.unwrap(); +// let client = FuelClient::from(srv.bound_address); +// +// // When +// let result = client.estimate_gas_price(10).await.unwrap(); +// +// // Then +// let actual = result.gas_price.0; +// assert_eq!(MIN_GAS_PRICE, actual) +// } // This test passed before this PR, but doesn't now #[tokio::test(flavor = "multi_thread")] @@ -464,13 +471,10 @@ async fn startup__can_override_gas_price_values_by_changing_config() { } use fuel_core_gas_price_service::v1::da_source_service::block_committer_costs::RawDaBlockCosts; +use fuel_core_storage::iter::IterDirection; #[test] fn produce_block__l1_committed_block_effects_gas_price() { - // let _ = tracing_subscriber::fmt() - // .with_max_level(tracing::Level::DEBUG) - // .try_init(); - let rt = tokio::runtime::Runtime::new().unwrap(); // set up chain with single unrecorded block let mut args = vec![ @@ -500,11 +504,12 @@ fn produce_block__l1_committed_block_effects_gas_price() { let mut mock = FakeServer::new(); let url = mock.url(); let costs = RawDaBlockCosts { - bundle_id: 1, - blocks_heights: vec![1], + id: 1, + start_height: 1, + end_height: 1, da_block_height: DaBlockHeight(100), - total_cost: 100, - total_size_bytes: 100, + cost: 100, + size: 100, }; mock.add_response(costs); @@ -672,19 +677,19 @@ fn _produce_block__algorithm_recovers_from_divergent_profit(block_delay: usize) }); let half_of_blocks = block_delay as u32 / 2; - let blocks_heights: Vec<_> = (1..half_of_blocks).collect(); - let count = blocks_heights.len() as u128; + let count = half_of_blocks; let block_bytes = 1000; let total_size_bytes = block_bytes * count as u32; let gas = 16 * total_size_bytes as u128; let cost_gwei = gas * 1; // blob gas price 1 gwei let cost = cost_gwei * 1_000_000_000; // Wei mock.add_response(RawDaBlockCosts { - bundle_id: 1, - blocks_heights, + id: 1, + start_height: 1, + end_height: half_of_blocks, da_block_height: DaBlockHeight(100), - total_cost: cost, - total_size_bytes, + cost, + size: total_size_bytes, }); let mut profits = Vec::new(); @@ -779,59 +784,24 @@ async fn produce_a_block(client: &FuelClient, rng: &mu let _ = client.produce_blocks(1, None).await.unwrap(); } -struct FakeServer { - server: mockito::ServerGuard, - responses: Arc, u32)>>, -} - -impl FakeServer { - fn new() -> Self { - let server = mockito::Server::new(); - let responses = Arc::new(Mutex::new((HashMap::new(), 0))); - let mut fake = Self { server, responses }; - fake.init(); - fake - } - - pub fn init(&mut self) { - let shared_responses = self.responses.clone(); - self.server - .mock("GET", Any) - .with_status(201) - .with_body_from_request(move |request| { - // take the requested number and return the corresponding response from the `responses` hashmap - let path = request.path(); - let maybe_sequence_number = - path.split('/').last().and_then(|x| x.parse::().ok()); - match maybe_sequence_number { - Some(sequence_number) => { - let guard = shared_responses.lock().unwrap(); - let responses = &guard.0; - let response = responses.get(&sequence_number).cloned(); - serde_json::to_string(&response).unwrap().into() - } - None => { - let guard = shared_responses.lock().unwrap(); - let responses = &guard.0; - let latest = &guard.1; - let response = responses.get(latest).cloned(); - serde_json::to_string(&response).unwrap().into() - } - } - }) - .expect_at_least(1) - .create(); - } - - pub fn add_response(&mut self, costs: RawDaBlockCosts) { - let mut guard = self.responses.lock().unwrap(); - let latest = guard.1; - let new_seq_no = latest + 1; - guard.0.insert(new_seq_no, costs); - guard.1 = new_seq_no; - } - - fn url(&self) -> String { - self.server.url() +#[test] +fn inspect_dbs() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .try_init(); + use fuel_core_storage::iter::IteratorOverTable; + + let db_path = "/Users/jamesturner/fuel/dev-net/.dev-net-db"; + let path = std::path::Path::new(db_path); + let db = + CombinedDatabase::open(path, 0, StateRewindPolicy::RewindFullRange, -1).unwrap(); + + let latest_recorded_blocks = db + .gas_price() + .iter_all::(Some(IterDirection::Reverse)); + tracing::info!("latest recorded blocks:"); + for block in latest_recorded_blocks { + let (_, recorded_height) = block.unwrap(); + tracing::info!("recorded height: {}", recorded_height); } }