fix(relayer): ensure local state is updated after logs downloaded (#2829)

## Linked Issues/PRs
<!-- List of related issues/PRs -->

## Description
<!-- List of detailed changes -->
delayed block production. includes deterministic start of block
production

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

### Before requesting review
- [ ] 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 <xgreenx9999@gmail.com>

Author: rymnc

.cargo/audit.toml

@@ -1,4 +1,5 @@
 [advisories]
 ignore = [
-    "RUSTSEC-2024-0421" # https://github.com/FuelLabs/fuel-core/issues/2488
-]
+    "RUSTSEC-2024-0421", # https://github.com/FuelLabs/fuel-core/issues/2488
+    "RUSTSEC-2025-0009", # https://github.com/FuelLabs/fuel-core/issues/2814
+]

crates/fuel-core/src/service.rs

@@ -3,6 +3,7 @@ use std::{
     sync::Arc,
 };
 
+use adapters::block_production_trigger::BlockProductionTrigger;
 pub use config::{
     Config,
     DbType,
@@ -74,11 +75,7 @@ pub struct SharedState {
     pub network: Option<fuel_core_p2p::service::SharedState>,
     #[cfg(feature = "relayer")]
     /// The Relayer shared state.
-    pub relayer: Option<
-        fuel_core_relayer::SharedState<
-            Database<crate::database::database_description::relayer::Relayer>,
-        >,
-    >,
+    pub relayer: Option<fuel_core_relayer::SharedState>,
     /// The GraphQL shared state.
     pub graph_ql: crate::fuel_core_graphql_api::api_service::SharedState,
     /// The underlying database.
@@ -138,11 +135,23 @@ impl FuelService {
         // initialize sub services
         tracing::info!("Initializing sub services");
         database.sync_aux_db_heights(shutdown_listener)?;
-        let (services, shared) = sub_services::init_sub_services(&config, database)?;
+
+        let block_production_trigger = BlockProductionTrigger::new();
+
+        let (services, shared) = sub_services::init_sub_services(
+            &config,
+            database,
+            block_production_trigger.clone(),
+        )?;
 
         let sub_services = Arc::new(services);
         let task = Task::new(sub_services.clone(), shared.clone())?;
-        let runner = ServiceRunner::new(task);
+        let runner = ServiceRunner::new_with_params(
+            task,
+            TaskParams {
+                block_production_trigger,
+            },
+        );
         let bound_address = runner.shared.graph_ql.bound_address;
 
         Ok(FuelService {
@@ -396,12 +405,17 @@ impl Task {
     }
 }
 
+#[derive(Default)]
+struct TaskParams {
+    block_production_trigger: BlockProductionTrigger,
+}
+
 #[async_trait::async_trait]
 impl RunnableService for Task {
     const NAME: &'static str = "FuelService";
     type SharedData = SharedState;
     type Task = Task;
-    type TaskParams = ();
+    type TaskParams = TaskParams;
 
     fn shared_data(&self) -> Self::SharedData {
         self.shared.clone()
@@ -410,7 +424,7 @@ impl RunnableService for Task {
     async fn into_task(
         mut self,
         watcher: &StateWatcher,
-        _: Self::TaskParams,
+        params: Self::TaskParams,
     ) -> anyhow::Result<Self::Task> {
         let mut watcher = watcher.clone();
 
@@ -424,6 +438,9 @@ impl RunnableService for Task {
                 }
             }
         }
+
+        params.block_production_trigger.send_trigger();
+
         Ok(self)
     }
 }

crates/fuel-core/src/service/adapters.rs

@@ -74,6 +74,8 @@ pub mod shared_sequencer;
 pub mod sync;
 pub mod txpool;
 
+pub mod block_production_trigger;
+
 #[derive(Debug, Clone)]
 pub struct ConsensusParametersProvider {
     shared_state: consensus_parameters_provider::SharedState,
@@ -407,7 +409,9 @@ impl ConsensusAdapter {
 #[derive(Clone)]
 pub struct MaybeRelayerAdapter {
     #[cfg(feature = "relayer")]
-    pub relayer_synced: Option<fuel_core_relayer::SharedState<Database<Relayer>>>,
+    pub relayer_synced: Option<fuel_core_relayer::SharedState>,
+    #[cfg(feature = "relayer")]
+    pub relayer_database: Database<Relayer>,
     #[cfg(feature = "relayer")]
     pub da_deploy_height: fuel_core_types::blockchain::primitives::DaBlockHeight,
 }

crates/fuel-core/src/service/adapters/block_production_trigger.rs

@@ -0,0 +1,31 @@
+use fuel_core_poa::ports::TriggerBlockProduction;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct BlockProductionTrigger {
+    notifier: Arc<tokio::sync::Notify>,
+}
+
+impl TriggerBlockProduction for BlockProductionTrigger {
+    async fn wait_for_trigger(&self) {
+        self.notifier.notified().await;
+    }
+}
+
+impl BlockProductionTrigger {
+    pub fn new() -> Self {
+        Self {
+            notifier: Arc::new(tokio::sync::Notify::new()),
+        }
+    }
+
+    pub fn send_trigger(&self) {
+        self.notifier.notify_one();
+    }
+}
+
+impl Default for BlockProductionTrigger {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/fuel-core/src/service/adapters/consensus_module.rs

@@ -67,7 +67,7 @@ impl RelayerPort for MaybeRelayerAdapter {
     ) -> anyhow::Result<()> {
         #[cfg(feature = "relayer")]
         {
-            if let Some(sync) = self.relayer_synced.as_ref() {
+            if let Some(sync) = &self.relayer_synced {
                 let current_height = sync.get_finalized_da_height();
                 anyhow::ensure!(
                     da_height.saturating_sub(*current_height) <= **_max_da_lag,

crates/fuel-core/src/service/adapters/producer.rs

@@ -132,7 +132,7 @@ impl fuel_core_producer::ports::Relayer for MaybeRelayerAdapter {
     ) -> anyhow::Result<DaBlockHeight> {
         #[cfg(feature = "relayer")]
         {
-            if let Some(sync) = self.relayer_synced.as_ref() {
+            if let Some(sync) = &self.relayer_synced {
                 sync.await_at_least_synced(height).await?;
                 let highest = sync.get_finalized_da_height();
                 Ok(highest)
@@ -157,14 +157,21 @@ impl fuel_core_producer::ports::Relayer for MaybeRelayerAdapter {
     ) -> anyhow::Result<RelayerBlockInfo> {
         #[cfg(feature = "relayer")]
         {
-            if let Some(sync) = self.relayer_synced.as_ref() {
-                get_gas_cost_and_transactions_number_for_height(**height, sync)
-            } else {
-                Ok(RelayerBlockInfo {
-                    gas_cost: 0,
-                    tx_count: 0,
-                })
-            }
+            let (gas_cost, tx_count) = self
+                .relayer_database
+                .get_events(height)?
+                .iter()
+                .fold((0u64, 0u64), |(gas_cost, tx_count), event| {
+                    let gas_cost = gas_cost.saturating_add(event.cost());
+                    let tx_count = match event {
+                        fuel_core_types::services::relayer::Event::Message(_) => tx_count,
+                        fuel_core_types::services::relayer::Event::Transaction(_) => {
+                            tx_count.saturating_add(1)
+                        }
+                    };
+                    (gas_cost, tx_count)
+                });
+            Ok(RelayerBlockInfo { gas_cost, tx_count })
         }
         #[cfg(not(feature = "relayer"))]
         {
@@ -181,35 +188,6 @@ impl fuel_core_producer::ports::Relayer for MaybeRelayerAdapter {
     }
 }
 
-#[cfg(feature = "relayer")]
-fn get_gas_cost_and_transactions_number_for_height(
-    height: u64,
-    sync: &fuel_core_relayer::SharedState<
-        crate::database::Database<
-            crate::database::database_description::relayer::Relayer,
-        >,
-    >,
-) -> anyhow::Result<RelayerBlockInfo> {
-    let da_height = DaBlockHeight(height);
-    let (gas_cost, tx_count) = sync
-        .database()
-        .storage::<fuel_core_relayer::storage::EventsHistory>()
-        .get(&da_height)?
-        .unwrap_or_default()
-        .iter()
-        .fold((0u64, 0u64), |(gas_cost, tx_count), event| {
-            let gas_cost = gas_cost.saturating_add(event.cost());
-            let tx_count = match event {
-                fuel_core_types::services::relayer::Event::Message(_) => tx_count,
-                fuel_core_types::services::relayer::Event::Transaction(_) => {
-                    tx_count.saturating_add(1)
-                }
-            };
-            (gas_cost, tx_count)
-        });
-    Ok(RelayerBlockInfo { gas_cost, tx_count })
-}
-
 impl fuel_core_producer::ports::BlockProducerDatabase for OnChainIterableKeyValueView {
     fn latest_height(&self) -> Option<BlockHeight> {
         self.latest_height().ok()

crates/fuel-core/src/service/adapters/relayer.rs

@@ -3,15 +3,26 @@ use crate::database::{
     Database,
 };
 use fuel_core_relayer::ports::Transactional;
-use fuel_core_storage::transactional::{
-    HistoricalView,
-    IntoTransaction,
-    StorageTransaction,
+use fuel_core_storage::{
+    transactional::{
+        AtomicView,
+        HistoricalView,
+        IntoTransaction,
+        StorageTransaction,
+    },
+    Result as StorageResult,
+    StorageAsRef,
+};
+use fuel_core_types::{
+    blockchain::primitives::DaBlockHeight,
+    services::relayer::Event,
 };
-use fuel_core_types::blockchain::primitives::DaBlockHeight;
 
 impl Transactional for Database<Relayer> {
-    type Transaction<'a> = StorageTransaction<&'a mut Self> where Self: 'a;
+    type Transaction<'a>
+        = StorageTransaction<&'a mut Self>
+    where
+        Self: 'a;
 
     fn transaction(&mut self) -> Self::Transaction<'_> {
         self.into_transaction()
@@ -21,3 +32,16 @@ impl Transactional for Database<Relayer> {
         HistoricalView::latest_height(self)
     }
 }
+
+impl Database<Relayer> {
+    pub fn get_events(&self, da_height: &DaBlockHeight) -> StorageResult<Vec<Event>> {
+        let events = self
+            .latest_view()?
+            .storage_as_ref::<fuel_core_relayer::storage::EventsHistory>()
+            .get(da_height)?
+            .unwrap_or_default()
+            .into_owned();
+
+        Ok(events)
+    }
+}

crates/fuel-core/src/service/sub_services.rs

@@ -17,6 +17,7 @@ use crate::{
     schema::build_schema,
     service::{
         adapters::{
+            block_production_trigger::BlockProductionTrigger,
             consensus_module::poa::InDirectoryPredefinedBlocks,
             consensus_parameters_provider,
             fuel_gas_price_provider::FuelGasPriceProvider,
@@ -66,6 +67,7 @@ pub type PoAService = fuel_core_poa::Service<
     SignMode,
     InDirectoryPredefinedBlocks,
     SystemTime,
+    BlockProductionTrigger,
 >;
 #[cfg(feature = "p2p")]
 pub type P2PService = fuel_core_p2p::service::Service<Database, TxPoolAdapter>;
@@ -86,6 +88,7 @@ pub const DEFAULT_GAS_PRICE_CHANGE_PERCENT: u16 = 10;
 pub fn init_sub_services(
     config: &Config,
     database: CombinedDatabase,
+    block_production_trigger: BlockProductionTrigger,
 ) -> anyhow::Result<(SubServices, SharedState)> {
     let chain_config = config.snapshot_reader.chain_config();
     let chain_id = chain_config.consensus_parameters.chain_id();
@@ -158,6 +161,8 @@ pub fn init_sub_services(
         #[cfg(feature = "relayer")]
         relayer_synced: relayer_service.as_ref().map(|r| r.shared.clone()),
         #[cfg(feature = "relayer")]
+        relayer_database: database.relayer().clone(),
+        #[cfg(feature = "relayer")]
         da_deploy_height: config.relayer.as_ref().map_or(
             DaBlockHeight(RelayerConfig::DEFAULT_DA_DEPLOY_HEIGHT),
             |config| config.da_deploy_height,
@@ -295,6 +300,7 @@ pub fn init_sub_services(
             signer,
             predefined_blocks,
             SystemTime,
+            block_production_trigger,
         )
     });
     let poa_adapter = PoAAdapter::new(poa.as_ref().map(|service| service.shared.clone()));
@@ -376,10 +382,6 @@ pub fn init_sub_services(
         Box::new(consensus_parameters_provider_service),
     ];
 
-    if let Some(poa) = poa {
-        services.push(Box::new(poa));
-    }
-
     #[cfg(feature = "relayer")]
     if let Some(relayer) = relayer_service {
         services.push(Box::new(relayer));
@@ -398,5 +400,10 @@ pub fn init_sub_services(
     services.push(Box::new(graph_ql));
     services.push(Box::new(graphql_worker));
 
+    // always make sure that the block producer is inserted last
+    if let Some(poa) = poa {
+        services.push(Box::new(poa));
+    }
+
     Ok((services, shared))
 }

crates/services/consensus_module/poa/src/ports.rs

@@ -140,3 +140,42 @@ impl PredefinedBlocks for InMemoryPredefinedBlocks {
 pub trait GetTime: Send + Sync {
     fn now(&self) -> Tai64;
 }
+
+pub trait TriggerBlockProduction: Send + Sync {
+    fn wait_for_trigger(&self) -> impl core::future::Future<Output = ()> + Send;
+}
+
+pub(crate) struct BlockProductionTrigger<TBP> {
+    notifier: TBP,
+    notified: std::sync::Arc<std::sync::atomic::AtomicBool>,
+}
+
+impl<TBP> TriggerBlockProduction for BlockProductionTrigger<TBP>
+where
+    TBP: TriggerBlockProduction,
+{
+    /// Cache the notification to avoid waiting for the trigger multiple times.
+    async fn wait_for_trigger(&self) {
+        if self.notified.load(std::sync::atomic::Ordering::Acquire) {
+            return;
+        }
+        self.notifier.wait_for_trigger().await;
+        // this is the first and only time we are notified
+        tracing::info!("Block Production has been triggered.");
+
+        self.notified
+            .store(true, std::sync::atomic::Ordering::Release);
+    }
+}
+
+impl<TBP> BlockProductionTrigger<TBP>
+where
+    TBP: TriggerBlockProduction,
+{
+    pub(crate) fn new(notifier: TBP) -> Self {
+        Self {
+            notifier,
+            notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        }
+    }
+}