fix(da_block_costs): remove Arc<Mutex<>> on shared_state and expose c…

…hannel (#2278)

## Linked Issues/PRs
<!-- List of related issues/PRs -->
- follows up #2192 with some
changes to clean up the da block costs source.

## Description
<!-- List of detailed changes -->
- Exposed a `broadcast::Sender` as the shared_state of the
DaBlockCostsSource, which is to be subscribed to when a service needs
this dependency.

## Checklist
- [x] Breaking changes are clearly marked as such in the PR description
and changelog
- [x] New behavior is reflected in tests
- [x] [The specification](https://github.com/FuelLabs/fuel-specs/)
matches the implemented behavior (link update PR if changes are needed)

### Before requesting review
- [x] I have reviewed the code myself
- [ ] I have created follow-up issues caused by this PR and linked them
here

### After merging, notify other teams

[Add or remove entries as needed]

- [ ] [Rust SDK](https://github.com/FuelLabs/fuels-rs/)
- [ ] [Sway compiler](https://github.com/FuelLabs/sway/)
- [ ] [Platform
documentation](https://github.com/FuelLabs/devrel-requests/issues/new?assignees=&labels=new+request&projects=&template=NEW-REQUEST.yml&title=%5BRequest%5D%3A+)
(for out-of-organization contributors, the person merging the PR will do
this)
- [ ] Someone else?

---------

Co-authored-by: Green Baneling <XgreenX9999@gmail.com>

crates/services/gas_price_service/src/v1.rs

@@ -1,2 +1,2 @@
 pub mod algorithm;
-pub mod da_source_adapter;
+pub mod da_source_service;

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

@@ -0,0 +1,98 @@
+use crate::v1::da_source_service::service::DaBlockCostsSource;
+use std::time::Duration;
+
+pub mod block_committer_costs;
+pub mod dummy_costs;
+pub mod service;
+
+#[derive(Debug, Default, Clone, Eq, Hash, PartialEq)]
+pub struct DaBlockCosts {
+    pub l2_block_range: core::ops::Range<u64>,
+    pub blob_size_bytes: u32,
+    pub blob_cost_wei: u128,
+}
+
+#[allow(non_snake_case)]
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::v1::da_source_service::{
+        dummy_costs::DummyDaBlockCosts,
+        service::new_service,
+    };
+    use fuel_core_services::Service;
+    use std::time::Duration;
+    use tokio::time::sleep;
+
+    #[tokio::test]
+    async fn run__when_da_block_cost_source_gives_value_shared_state_is_updated() {
+        // given
+        let expected_da_cost = DaBlockCosts {
+            l2_block_range: 0..10,
+            blob_size_bytes: 1024 * 128,
+            blob_cost_wei: 2,
+        };
+        let da_block_costs_source = DummyDaBlockCosts::new(Ok(expected_da_cost.clone()));
+        let service = new_service(da_block_costs_source, Some(Duration::from_millis(1)));
+        let mut shared_state = &mut service.shared.subscribe();
+
+        // when
+        service.start_and_await().await.unwrap();
+        sleep(Duration::from_millis(10)).await;
+        service.stop_and_await().await.unwrap();
+
+        // then
+        let da_block_costs = shared_state.try_recv().unwrap();
+        assert_eq!(da_block_costs, expected_da_cost);
+    }
+
+    #[tokio::test]
+    async fn run__when_da_block_cost_source_gives_value_shared_state_is_marked_stale() {
+        // given
+        let expected_da_cost = DaBlockCosts {
+            l2_block_range: 0..10,
+            blob_size_bytes: 1024 * 128,
+            blob_cost_wei: 1,
+        };
+        let da_block_costs_source = DummyDaBlockCosts::new(Ok(expected_da_cost.clone()));
+        let service = new_service(da_block_costs_source, Some(Duration::from_millis(8)));
+        let mut shared_state = &mut service.shared.subscribe();
+
+        // when
+        service.start_and_await().await.unwrap();
+        sleep(Duration::from_millis(10)).await;
+        service.stop_and_await().await.unwrap();
+
+        let actual = shared_state.try_recv().unwrap();
+        assert_eq!(actual, expected_da_cost);
+
+        // then
+        let da_block_costs_res = shared_state.try_recv();
+        assert!(da_block_costs_res.is_err());
+        assert!(matches!(
+            da_block_costs_res.err().unwrap(),
+            tokio::sync::broadcast::error::TryRecvError::Empty
+        ));
+    }
+
+    #[tokio::test]
+    async fn run__when_da_block_cost_source_errors_shared_state_is_not_updated() {
+        // given
+        let da_block_costs_source = DummyDaBlockCosts::new(Err(anyhow::anyhow!("boo!")));
+        let service = new_service(da_block_costs_source, Some(Duration::from_millis(1)));
+        let mut shared_state = &mut service.shared.subscribe();
+
+        // when
+        service.start_and_await().await.unwrap();
+        sleep(Duration::from_millis(10)).await;
+        service.stop_and_await().await.unwrap();
+
+        // then
+        let da_block_costs_res = shared_state.try_recv();
+        assert!(da_block_costs_res.is_err());
+        assert!(matches!(
+            da_block_costs_res.err().unwrap(),
+            tokio::sync::broadcast::error::TryRecvError::Empty
+        ));
+    }
+}

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

@@ -1,6 +1,6 @@
 #![allow(clippy::arithmetic_side_effects)]
 
-use crate::v1::da_source_adapter::{
+use crate::v1::da_source_service::{
     service::{
         DaBlockCostsSource,
         Result as DaBlockCostsResult,

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

@@ -1,4 +1,4 @@
-use crate::v1::da_source_adapter::{
+use crate::v1::da_source_service::{
     service::{
         DaBlockCostsSource,
         Result as DaBlockCostsResult,