fuel-core-sync (#889)

- Closes #881 
- Closes #882 
- Closes #880

Co-authored-by: green <xgreenx9999@gmail.com>

.gitignore

@@ -6,4 +6,5 @@
 .vscode
 .cov
 lcov.info
-version-compatibility/Cargo.lock
+version-compatibility/Cargo.lock
+benches/benches-outputs/Cargo.lock

crates/services/Cargo.toml

@@ -1,18 +1,19 @@
 [package]
-name = "fuel-core-services"
-version = "0.15.1"
 authors = ["Fuel Labs <contact@fuel.sh>"]
+description = "The common code for fuel core services."
 edition = "2021"
 homepage = "https://fuel.network/"
-keywords = ["blockchain", "fuel", "consensus", "bft"]
+keywords = ["bft", "blockchain", "consensus", "fuel"]
 license = "BUSL-1.1"
+name = "fuel-core-services"
 repository = "https://github.com/FuelLabs/fuel-core"
-description = "The common code for fuel core services."
+version = "0.15.1"
 
 [dependencies]
 anyhow = "1.0"
 async-trait = "0.1"
 futures = "0.3"
+parking_lot = "0.12"
 tokio = { version = "1.21", features = ["full"] }
 tracing = "0.1"
 

crates/services/src/lib.rs

@@ -17,6 +17,19 @@ pub mod stream {
     /// A Send + Sync BoxStream
     pub type BoxStream<T> =
         core::pin::Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>;
+
+    /// Helper trait to create a BoxStream from a Stream
+    pub trait IntoBoxStream: Stream {
+        /// Convert this stream into a BoxStream.
+        fn into_boxed(self) -> BoxStream<Self::Item>
+        where
+            Self: Sized + Send + Sync + 'static,
+        {
+            Box::pin(self)
+        }
+    }
+
+    impl<S> IntoBoxStream for S where S: Stream + Send + Sync + 'static {}
 }
 
 pub use service::{
@@ -26,6 +39,7 @@ pub use service::{
     Service,
     ServiceRunner,
     Shared,
+    SharedMutex,
 };
 pub use state::{
     State,

crates/services/src/service.rs

@@ -11,6 +11,16 @@ use tokio::{
 /// Alias for Arc<T>
 pub type Shared<T> = std::sync::Arc<T>;
 
+/// A mutex that can safely be in async contexts and avoids deadlocks.
+#[derive(Debug)]
+pub struct SharedMutex<T>(Shared<parking_lot::Mutex<T>>);
+
+impl<T> Clone for SharedMutex<T> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
 /// Used if services have no asynchronously shared data
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct EmptyShared;
@@ -290,6 +300,19 @@ where
     })
 }
 
+impl<T> SharedMutex<T> {
+    /// Creates a new `SharedMutex` with the given value.
+    pub fn new(t: T) -> Self {
+        Self(Shared::new(parking_lot::Mutex::new(t)))
+    }
+
+    /// Apply a function to the inner value and return a value.
+    pub fn apply<R>(&self, f: impl FnOnce(&mut T) -> R) -> R {
+        let mut t = self.0.lock();
+        f(&mut t)
+    }
+}
+
 // TODO: Add tests
 #[cfg(test)]
 mod tests {

crates/services/src/state.rs

@@ -76,6 +76,10 @@ impl StateWatcher {
 impl StateWatcher {
     /// Infinity loop while the state is `State::Started`. Returns the next received state.
     pub async fn while_started(&mut self) -> anyhow::Result<State> {
+        let state = self.borrow().clone();
+        if !state.started() {
+            return Ok(state)
+        }
         loop {
             self.changed().await?;
 

crates/services/sync/Cargo.toml

@@ -1,16 +1,25 @@
 [package]
-name = "fuel-core-sync"
-version = "0.15.1"
 authors = ["Fuel Labs <contact@fuel.sh>"]
+description = "Fuel Synchronizer"
 edition = "2021"
 homepage = "https://fuel.network/"
 keywords = ["blockchain", "fuel", "fuel-vm"]
 license = "BUSL-1.1"
+name = "fuel-core-sync"
 repository = "https://github.com/FuelLabs/fuel-core"
-description = "Fuel Synchronizer"
+version = "0.15.1"
 
 [dependencies]
 anyhow = "1.0"
+async-trait = "0.1.60"
+fuel-core-services = { path = "../" }
 fuel-core-types = { path = "../../types", version = "0.15.1" }
-parking_lot = "0.12"
+futures = "0.3.25"
 tokio = { version = "1.21", features = ["full"] }
+
+[dev-dependencies]
+fuel-core-types = { path = "../../types", features = [
+    "test-helpers",
+] }
+mockall = "0.11.3"
+test-case = "2.2.2"