chore(runnable_task): remove async_trait usage (#2660)

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

## Description
<!-- List of detailed changes -->

should improve performance. breaking changes for anyone using fuel-core
as a dependency, more specifically the `RunnableTask` trait. maybe cc:
@FuelLabs/data-systems ?

## 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
- [x] 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?

crates/fuel-core/src/graphql_api/api_service.rs

@@ -193,7 +193,6 @@ impl RunnableService for GraphqlService {
     }
 }
 
-#[async_trait::async_trait]
 impl RunnableTask for Task {
     async fn run(&mut self, _: &mut StateWatcher) -> TaskNextAction {
         match self.server.as_mut().await {

crates/fuel-core/src/graphql_api/worker_service.rs

@@ -669,7 +669,6 @@ where
     Ok(())
 }
 
-#[async_trait::async_trait]
 impl<TxPool, D> RunnableTask for Task<TxPool, D>
 where
     TxPool: ports::worker::TxPool,

crates/fuel-core/src/service.rs

@@ -428,7 +428,6 @@ impl RunnableService for Task {
     }
 }
 
-#[async_trait::async_trait]
 impl RunnableTask for Task {
     #[tracing::instrument(skip_all)]
     async fn run(&mut self, watcher: &mut StateWatcher) -> TaskNextAction {

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

@@ -110,7 +110,6 @@ impl SharedState {
     }
 }
 
-#[async_trait::async_trait]
 impl RunnableTask for Task {
     async fn run(&mut self, watcher: &mut StateWatcher) -> TaskNextAction {
         tokio::select! {

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

@@ -506,7 +506,6 @@ where
     }
 }
 
-#[async_trait::async_trait]
 impl<T, B, I, S, PB, C> RunnableTask for MainTask<T, B, I, S, PB, C>
 where
     T: TransactionPool,

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

@@ -136,7 +136,6 @@ impl RunnableService for SyncTask {
     }
 }
 
-#[async_trait::async_trait]
 impl RunnableTask for SyncTask {
     async fn run(&mut self, watcher: &mut StateWatcher) -> TaskNextAction {
         let tick: BoxFuture<tokio::time::Instant> = if let Some(timer) = &mut self.timer {
@@ -329,7 +328,7 @@ mod tests {
         }
     }
 
-    /// Helper function that creates a `SyncTask` with a given configuration    
+    /// Helper function that creates a `SyncTask` with a given configuration
     fn configure_sync_task(
         min_connected_reserved_peers: usize,
         connections_stream: impl IntoIterator<Item = usize>,

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

@@ -11,7 +11,6 @@ use crate::{
     v0::algorithm::SharedV0Algorithm,
 };
 use anyhow::anyhow;
-use async_trait::async_trait;
 use fuel_core_services::{
     RunnableTask,
     StateWatcher,
@@ -137,7 +136,6 @@ where
         Ok(())
     }
 }
-#[async_trait]
 impl<L2, Metadata> RunnableTask for GasPriceServiceV0<L2, Metadata>
 where
     L2: L2BlockSource,

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

@@ -181,7 +181,6 @@ where
     }
 }
 
-#[async_trait::async_trait]
 impl<Source> RunnableTask for DaSourceService<Source>
 where
     Source: DaBlockCostsSource,

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

@@ -406,7 +406,6 @@ where
     }
 }
 
-#[async_trait]
 impl<L2, DA, AtomicStorage> RunnableTask for GasPriceServiceV1<L2, DA, AtomicStorage>
 where
     L2: L2BlockSource,

crates/services/p2p/src/service.rs

@@ -869,7 +869,6 @@ where
 }
 
 // TODO: Add tests https://github.com/FuelLabs/fuel-core/issues/1275
-#[async_trait::async_trait]
 impl<P, V, B, T> RunnableTask for Task<P, V, B, T>
 where
     P: TaskP2PService + 'static,
@@ -1019,8 +1018,14 @@ where
                         tracing::error!("Failed to perform peer heartbeat reputation checks: {:?}", e);
                     }
                 }
-                self.next_check_time += self.heartbeat_check_interval;
-                TaskNextAction::Continue
+
+                if let Some(next_check_time) = self.next_check_time.checked_add(self.heartbeat_check_interval) {
+                    self.next_check_time = next_check_time;
+                    TaskNextAction::Continue
+                } else {
+                    tracing::error!("Next check time overflowed");
+                    TaskNextAction::Stop
+                }
             }
         }
     }

crates/services/relayer/src/service.rs

@@ -213,7 +213,6 @@ where
     }
 }
 
-#[async_trait]
 impl<P, D> RunnableTask for Task<P, D>
 where
     P: Middleware<Error = ProviderError> + 'static,

crates/services/shared-sequencer/src/service.rs

@@ -169,7 +169,6 @@ where
     }
 }
 
-#[async_trait]
 impl<S> RunnableTask for Task<S>
 where
     S: Signer + 'static,

crates/services/src/service.rs

@@ -123,7 +123,6 @@ impl From<Result<bool, anyhow::Error>> for TaskNextAction {
 
 /// The trait is implemented by the service task and contains a single iteration of the infinity
 /// loop.
-#[async_trait::async_trait]
 pub trait RunnableTask: Send {
     /// This function should contain the main business logic of the service task. It will run until
     /// the service either returns false, panics or a stop signal is received.
@@ -134,10 +133,13 @@ pub trait RunnableTask: Send {
     /// `State::Started`. So first, the `run` method should return a value, and after, the service
     /// will stop. If the service should react to the state change earlier, it should handle it in
     /// the `run` loop on its own. See [`StateWatcher::while_started`].
-    async fn run(&mut self, watcher: &mut StateWatcher) -> TaskNextAction;
+    fn run(
+        &mut self,
+        watcher: &mut StateWatcher,
+    ) -> impl core::future::Future<Output = TaskNextAction> + Send;
 
     /// Gracefully shutdowns the task after the end of the execution cycle.
-    async fn shutdown(self) -> anyhow::Result<()>;
+    fn shutdown(self) -> impl core::future::Future<Output = anyhow::Result<()>> + Send;
 }
 
 /// The service runner manages the lifecycle, execution and error handling of a `RunnableService`.
@@ -458,7 +460,6 @@ fn panic_to_string(e: Box<dyn core::any::Any + Send>) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures::future::BoxFuture;
 
     mockall::mock! {
         Service {}
@@ -480,16 +481,11 @@ mod tests {
     mockall::mock! {
         Task {}
 
-        #[async_trait::async_trait]
         impl RunnableTask for Task {
-            fn run<'_self, '_state, 'a>(
-                &'_self mut self,
-                state: &'_state mut StateWatcher
-            ) -> BoxFuture<'a, TaskNextAction>
-            where
-                '_self: 'a,
-                '_state: 'a,
-                Self: Sync + 'a;
+            fn run(
+                &mut self,
+                state: &mut StateWatcher
+            ) -> impl core::future::Future<Output = TaskNextAction> + Send;
 
             async fn shutdown(self) -> anyhow::Result<()>;
         }

crates/services/sync/src/service.rs

@@ -112,7 +112,6 @@ where
     }
 }
 
-#[async_trait::async_trait]
 impl<P, E, C> RunnableTask for SyncTask<P, E, C>
 where
     P: PeerToPeerPort + Send + Sync + 'static,
@@ -171,7 +170,6 @@ where
     }
 }
 
-#[async_trait::async_trait]
 impl<P, E, C> RunnableTask for ImportTask<P, E, C>
 where
     P: PeerToPeerPort + Send + Sync + 'static,

crates/services/txpool_v2/src/service.rs

@@ -228,7 +228,6 @@ where
     }
 }
 
-#[async_trait::async_trait]
 impl<View> RunnableTask for Task<View>
 where
     View: TxPoolPersistentStorage,