Merge pull request #103 from chrivers/chrivers/svc-crate

Implement a new crate ("svc") to manage rust-based micro-services.

Bifrost consists of a number of "things" running independently, while performing work either on request, on a timer, or some combination of the two.

Synchronizing, coordinating and sharing information between all these various parts of Bifrost, takes up a non-trivial amount of the codebase.

That's why we are introducing a dedicates library (a "crate", in rust terminology) for managing services that implement a standardized interface, specifically, the `Service` trait.

This makes it possible to share start/run/stop handling for all services, as well as more advanced features like policies ("if a service fails, what should happen?"), error handling, dependencies, and liveness checks.

This PR introduces and builds up the "svc" crate, but does not change Bifrost to use it, yet.

Author: chrivers

crates/svc/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "svc"
+version = "0.1.0"
+edition.workspace = true
+authors.workspace = true
+rust-version.workspace = true
+description.workspace = true
+readme.workspace = true
+repository.workspace = true
+license.workspace = true
+categories.workspace = true
+keywords.workspace = true
+
+[dependencies]
+async-trait = "0.1.86"
+futures = { version = "0.3.31", default-features = false, features = ["alloc"] }
+log = "0.4.26"
+serde = { version = "1.0.218", features = ["derive"] }
+thiserror = "2.0.11"
+tokio = { version = "1.43.0", features = ["io-util", "macros", "process", "rt", "rt-multi-thread", "sync", "time", "tokio-macros"] }
+uuid = { version = "1.14.0", features = ["v4"] }
+
+[lints]
+workspace = true
+
+[dev-dependencies]
+pretty_env_logger = "0.5.0"

crates/svc/examples/from_async_fn.rs

@@ -0,0 +1,50 @@
+use std::time::Duration;
+
+use thiserror::Error;
+use tokio::time::sleep;
+
+use svc::error::SvcResult;
+use svc::manager::ServiceManager;
+
+#[derive(Error, Debug)]
+pub enum SimpleError {
+    #[error("That didn't work")]
+    Nope,
+}
+
+async fn run() -> Result<(), SimpleError> {
+    let dur = Duration::from_millis(800);
+
+    println!("Hello");
+
+    println!("1");
+    sleep(dur).await;
+    println!("2");
+    sleep(dur).await;
+    println!("3");
+
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> SvcResult<()> {
+    pretty_env_logger::formatted_builder()
+        .filter_level(log::LevelFilter::Debug)
+        .parse_default_env()
+        .init();
+
+    let (mut client, future) = ServiceManager::spawn();
+
+    client.register_function("foo", run()).await?;
+    client.start("foo").await?;
+    println!("main: service configured");
+
+    client.wait_for_start("foo").await?;
+    println!("main: service started");
+
+    client.shutdown().await?;
+    future.await??;
+    println!("main: service stopped");
+
+    Ok(())
+}

crates/svc/examples/policy.rs

@@ -0,0 +1,70 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+use svc::policy::{Policy, Retry};
+use svc::runservice::StandardService;
+use thiserror::Error;
+
+use svc::error::SvcResult;
+use svc::manager::ServiceManager;
+use svc::traits::{Service, ServiceState};
+
+#[derive(Clone)]
+struct PolicyService {
+    counter: u32,
+}
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("Not done yet")]
+    MoreToDo,
+}
+
+#[async_trait]
+impl Service for PolicyService {
+    type Error = Error;
+
+    async fn run(&mut self) -> Result<(), Error> {
+        println!("Hello {}", self.counter);
+        self.counter += 1;
+
+        // returning an Err will invoke the policy for the Running state
+        Err(Error::MoreToDo)
+    }
+}
+
+#[tokio::main]
+async fn main() -> SvcResult<()> {
+    const NAME: &str = "policy-service";
+
+    pretty_env_logger::formatted_builder()
+        .filter_level(log::LevelFilter::Debug)
+        .parse_default_env()
+        .init();
+
+    let (mut client, future) = ServiceManager::spawn();
+
+    let svc = PolicyService { counter: 0 };
+
+    // Manually construct a ServiceRunner, and set a specific policy for
+    // handling errors during .run()
+    let svcr = StandardService::new(NAME, svc).with_run_policy(
+        // Try up to 5 times, waiting 300ms between each attempt
+        Policy::new()
+            .with_retry(Retry::Limit(5))
+            .with_delay(Duration::from_millis(300)),
+    );
+
+    let uuid = client.register(NAME, svcr).await?;
+    client.start(uuid).await?;
+    println!("main: service will attempt to run 5 times");
+
+    client.wait_for_start(uuid).await?;
+    println!("main: service started");
+
+    client.wait_for_state(uuid, ServiceState::Failed).await?;
+    client.shutdown().await?;
+    future.await??;
+
+    Ok(())
+}

crates/svc/examples/restart.rs

@@ -0,0 +1,56 @@
+use std::time::Duration;
+
+use tokio::time::sleep;
+
+use svc::error::{RunSvcError, SvcResult};
+use svc::manager::ServiceManager;
+
+async fn run() -> Result<(), RunSvcError> {
+    let dur = Duration::from_millis(200);
+
+    println!("Hello");
+
+    let mut counter = 0;
+
+    loop {
+        println!("{counter}");
+        sleep(dur).await;
+        counter += 1;
+    }
+}
+
+#[tokio::main]
+async fn main() -> SvcResult<()> {
+    pretty_env_logger::formatted_builder()
+        .filter_level(log::LevelFilter::Debug)
+        .parse_default_env()
+        .init();
+
+    let (mut client, future) = ServiceManager::spawn();
+
+    client.register_function("foo", run()).await?;
+
+    client.start("foo").await?;
+    println!("main: service configured");
+
+    client.wait_for_start("foo").await?;
+    println!("main: service started");
+
+    sleep(Duration::from_millis(1000)).await;
+
+    client.stop("foo").await?;
+    client.wait_for_stop("foo").await?;
+
+    println!("main: service stopped");
+
+    client.start("foo").await?;
+    client.wait_for_start("foo").await?;
+    println!("main: service started");
+
+    sleep(Duration::from_millis(1000)).await;
+    client.shutdown().await?;
+
+    future.await??;
+
+    Ok(())
+}

crates/svc/examples/simple.rs

@@ -0,0 +1,86 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+use thiserror::Error;
+use tokio::time::sleep;
+
+use svc::error::SvcResult;
+use svc::manager::ServiceManager;
+use svc::traits::Service;
+
+#[derive(Clone)]
+struct Simple {
+    name: String,
+    counter: u32,
+}
+
+#[derive(Error, Debug)]
+pub enum SimpleError {
+    #[error("That didn't work..")]
+    Nope,
+}
+
+#[async_trait]
+impl Service for Simple {
+    type Error = SimpleError;
+
+    async fn run(&mut self) -> Result<(), SimpleError> {
+        let dur = Duration::from_millis(300);
+
+        println!("Hello from {}", self.name);
+
+        println!("1");
+        sleep(dur).await;
+        println!("2");
+        sleep(dur).await;
+        println!("3");
+
+        println!("Done running. Now going to stop (this will fail the first time)");
+        Ok(())
+    }
+
+    async fn stop(&mut self) -> Result<(), SimpleError> {
+        self.counter += 1;
+
+        // pretend this service doesn't succeed at stopping right away
+        if self.counter > 1 {
+            Ok(())
+        } else {
+            Err(SimpleError::Nope)
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> SvcResult<()> {
+    pretty_env_logger::formatted_builder()
+        .filter_level(log::LevelFilter::Debug)
+        .parse_default_env()
+        .init();
+
+    let (mut client, future) = ServiceManager::spawn();
+
+    let svc = Simple {
+        name: "Simple Service".to_string(),
+        counter: 0,
+    };
+
+    client.register_service("foo", svc).await?;
+    client.start("foo").await?;
+
+    println!("main: service configured");
+
+    client.wait_for_start("foo").await?;
+
+    println!("main: service started");
+
+    client.wait_for_stop("foo").await?;
+
+    println!("main: service stopped");
+
+    client.shutdown().await?;
+
+    future.await??;
+
+    Ok(())
+}

crates/svc/src/error.rs

@@ -0,0 +1,75 @@
+use std::error::Error;
+
+use thiserror::Error;
+
+use crate::manager::{ServiceEvent, SvmRequest};
+use crate::serviceid::ServiceId;
+use crate::traits::ServiceState;
+
+#[derive(Error, Debug)]
+pub enum SvcError {
+    /* mapped errors */
+    #[error(transparent)]
+    FromUtf8Error(#[from] std::string::FromUtf8Error),
+
+    #[error(transparent)]
+    IOError(#[from] std::io::Error),
+
+    #[error(transparent)]
+    TryFromIntError(#[from] std::num::TryFromIntError),
+
+    #[error(transparent)]
+    UuidError(#[from] uuid::Error),
+
+    #[error(transparent)]
+    JoinError(#[from] tokio::task::JoinError),
+
+    #[error(transparent)]
+    MpscSendError(#[from] tokio::sync::mpsc::error::SendError<SvmRequest>),
+
+    #[error(transparent)]
+    MpscSendEventError(#[from] tokio::sync::mpsc::error::SendError<ServiceEvent>),
+
+    #[error(transparent)]
+    WatchSendError(#[from] tokio::sync::watch::error::SendError<ServiceState>),
+
+    #[error(transparent)]
+    WatchRecvError(#[from] tokio::sync::watch::error::RecvError),
+
+    #[error(transparent)]
+    OneshotRecvError(#[from] tokio::sync::oneshot::error::RecvError),
+
+    #[error("Service {0:?} not registered")]
+    ServiceNotFound(ServiceId),
+
+    #[error("Service {0} already exists")]
+    ServiceAlreadyExists(String),
+
+    #[error("All services stopped")]
+    Shutdown,
+
+    #[error("Service has failed")]
+    ServiceFailed,
+}
+
+#[derive(Error, Debug)]
+pub enum RunSvcError {
+    /* mapped errors */
+    #[error(transparent)]
+    MpscSendError(#[from] tokio::sync::mpsc::error::SendError<SvmRequest>),
+
+    #[error(transparent)]
+    WatchSendError(#[from] tokio::sync::watch::error::SendError<ServiceState>),
+
+    #[error(transparent)]
+    MpscSendEventError(#[from] tokio::sync::mpsc::error::SendError<ServiceEvent>),
+
+    #[error(transparent)]
+    WatchRecvError(#[from] tokio::sync::watch::error::RecvError),
+
+    /* errors from run service */
+    #[error(transparent)]
+    ServiceError(Box<dyn Error + Send>),
+}
+
+pub type SvcResult<T> = Result<T, SvcError>;

crates/svc/src/lib.rs

@@ -0,0 +1,7 @@
+pub mod error;
+pub mod manager;
+pub mod policy;
+pub mod rpc;
+pub mod runservice;
+pub mod serviceid;
+pub mod traits;