Merge pull request #25 from balena-io-experimental/improved-docs

Write documentation

Author: flowzone-app[bot]

Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "gustav"
+name = "mahler"
 version = "0.14.0"
 edition = "2021"
 description = "An automated job orchestration library to build and execute dynamic workflows"
@@ -14,7 +14,7 @@ anyhow = "1.0.97"
 json-patch = "4"
 jsonptr = "0.7.1"
 matchit = "0.8.4"
-serde = { version = "1.0.197" }
+serde = "1.0.197"
 serde_json = "1.0.120"
 thiserror = "2"
 tokio = { version = "1.43.0", default-features = false, features = [
@@ -36,6 +36,8 @@ tracing-subscriber = { version = "0.3.19", default-features = false, optional =
 ] }
 
 [dev-dependencies]
+# configure mahler with the logging feature to ensure documentation tests pass
+mahler = { path = ".", features = ["logging"] }
 tracing-subscriber = { version = "0.3.19", default-features = false, features = [
   "ansi",
   "registry",
@@ -45,6 +47,13 @@ tracing-subscriber = { version = "0.3.19", default-features = false, features =
 dedent = "0.1.1"
 env_logger = "0.11.8"
 log = "0.4.25"
-tokio = { version = "1.36.0", features = ["rt", "time"] }
+tokio = { version = "1.36.0", features = ["rt-multi-thread", "time"] }
 pretty_assertions = "1.4.1"
 tokio-test = "0.4.4"
+
+# docs.rs-specific configuration
+[package.metadata.docs.rs]
+# document all features
+all-features = true
+# defines the configuration attribute `docsrs`
+rustdoc-args = ["--cfg", "docsrs"]

README.md

@@ -1,5 +1,5 @@
 [package]
-name = "gustav-examples-composer"
+name = "mahler-examples-composer"
 version = "0.0.0"
 edition = "2021"
 publish = false
@@ -8,7 +8,7 @@ publish = false
 anyhow = "1.0.98"
 bollard = { version = "0.18.1" }
 futures-util = { version = "0.3.31", default-features = false }
-gustav = { path = "../../", features = ["logging"] }
+mahler = { path = "../../", features = ["logging"] }
 log = "0.4.27"
 serde = { version = "1.0.219", features = ["derive"] }
 thiserror = "2.0.12"

examples/composer/Cargo.toml

@@ -0,0 +1,5 @@
+# Container composition example
+
+Use mahler to manage container composition.
+
+See the [library tests](https://github.com/balena-io-experimental/mahler-rs/blob/769435795c9bef779adab53efcb51812c34a0296/examples/composer/src/lib.rs#L770) for usage examples

examples/composer/README.md

@@ -2,16 +2,16 @@ use anyhow::{anyhow, Context as AnyhowCtx};
 use bollard::container::{CreateContainerOptions, StartContainerOptions};
 use bollard::secret::{ContainerInspectResponse, ContainerState, ContainerStateStatusEnum};
 use futures_util::stream::StreamExt;
-use gustav::worker::{Ready, Worker};
+use mahler::worker::{Ready, Worker};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use thiserror::Error;
 
 use bollard::image::CreateImageOptions;
 use bollard::Docker;
 
-use gustav::extract::{Args, Pointer, Res, System, Target, View};
-use gustav::task::prelude::*;
+use mahler::extract::{Args, Pointer, Res, System, Target, View};
+use mahler::task::prelude::*;
 
 #[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone, Default)]
 pub enum ServiceStatus {
@@ -672,8 +672,8 @@ pub fn create_worker(project: Project) -> Worker<Project, Ready, TargetProject>
 #[cfg(test)]
 mod tests {
     use bollard::container::{ListContainersOptions, RemoveContainerOptions};
-    use gustav::worker::SeekStatus;
-    use gustav::{seq, Dag};
+    use mahler::worker::SeekStatus;
+    use mahler::{seq, Dag};
     use pretty_assertions::assert_eq;
     use serde_json::json;
     use tracing_subscriber::fmt::format::FmtSpan;
@@ -785,7 +785,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target.clone()).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetReached);
+        assert_eq!(worker.status(), &SeekStatus::Success);
 
         // The alpine image must exist now
         let docker = Docker::connect_with_defaults().unwrap();
@@ -943,7 +943,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetReached);
+        assert_eq!(worker.status(), &SeekStatus::Success);
 
         // The alpine image must exist now
         let docker = Docker::connect_with_defaults().unwrap();
@@ -982,7 +982,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetReached);
+        assert_eq!(worker.status(), &SeekStatus::Success);
 
         // The container ids should match
         let container = docker
@@ -1009,7 +1009,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetReached);
+        assert_eq!(worker.status(), &SeekStatus::Success);
 
         // The container ids should match
         let container = docker
@@ -1030,7 +1030,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetReached);
+        assert_eq!(worker.status(), &SeekStatus::Success);
 
         // The container should no longer exist
         let container = docker

examples/composer/src/lib.rs

@@ -0,0 +1,20 @@
+[package]
+name = "mahler-examples-readme"
+version = "0.0.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+anyhow = "1.0.98"
+env_logger = "0.11.8"
+log = "0.4.27"
+mahler = { path = "../../", features = ["logging"] }
+serde = "1.0.219"
+tokio = { version = "1.44.2", default-features = false, features = [
+  "rt-multi-thread",
+  "time",
+] }
+
+[dev-dependencies]
+assert_cmd = "2.0.17"
+predicates = "3.1.3"

examples/readme/Cargo.toml

@@ -0,0 +1,108 @@
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use std::time::Duration;
+use tokio::time::sleep;
+
+use mahler::extract::{Args, Target, View};
+use mahler::task::prelude::*;
+use mahler::worker::{init_logging, Worker};
+
+// The state model needs to be Serializable and Deserializable
+// since the library uses JSON internally to access parts
+// of the state
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
+struct Counters(BTreeMap<String, i32>);
+
+// `plus_one` defines a job that updates a counter if it is below some target.
+// The job makes use of two extractors:
+// - `View`, that provides a mutable view into the system state. By modifying the view,
+// the job task can affect the global state
+// - `Target`, providing a read only view to the target being seeked by the planner
+fn plus_one(mut counter: View<i32>, Target(tgt): Target<i32>) -> Update<i32> {
+    if *counter < tgt {
+        // Modify the counter value if we are below the target
+        *counter += 1;
+    }
+
+    // The task is called at planning and at runtime, the `with_io` function
+    // allows us to define what is returned by the function at each context.
+    // The first argument of the function is what the planner receives,
+    // the right side of the call is what will be executed at runtime if the
+    // task is selected.
+    with_io(counter, |counter| async {
+        // The async call can be used to actually make changes to the underlying system.
+        // It could be writing the counter to a database or a file. In this
+        // case we just add some timer
+        sleep(Duration::from_millis(10)).await;
+        Ok(counter)
+    })
+}
+
+// `plus_two` is a compound job. Compound job do not modify the state directly
+// but return combination of sub-tasks that are applicable to a certain target
+fn plus_two(counter: View<i32>, Target(tgt): Target<i32>) -> Vec<Task> {
+    // If the difference between the current state and target is >1
+    if tgt - *counter > 1 {
+        // Then return a sequence of two tasks with the same target
+        return vec![plus_one.with_target(tgt), plus_one.with_target(tgt)];
+    }
+
+    // Otherwise do nothing
+    vec![]
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Convert tracing logs into `log` crate logs
+    init_logging();
+
+    // Show logs on stdout
+    env_logger::builder().format_target(false).init();
+
+    let worker = Worker::new()
+        // The jobs are applicable to `UPDATE` operations
+        // on individual counters
+        .job(
+            "/{counter}",
+            update(plus_one)
+                // we can add a description to the job for the logs
+                .with_description(|Args(counter): Args<String>| format!("{counter}++")),
+        )
+        .job("/{counter}", update(plus_two))
+        // We initialize the worker with two counters
+        // `a` and `b` with value 0
+        .initial_state(Counters(BTreeMap::from([
+            ("a".to_string(), 0),
+            ("b".to_string(), 0),
+        ])))
+        .with_context(|| "failed to serialize initial state")?;
+
+    // Tell the worker to find a plan from the initial state (a:0, b:0)
+    // to the target state (a:1, b:2) and execute it
+    let worker = worker
+        .seek_target(Counters(BTreeMap::from([
+            ("a".to_string(), 1),
+            ("b".to_string(), 2),
+        ])))
+        .await
+        .with_context(|| "failed to reach target state")?;
+
+    // Get the internal state from the Worker. The worker
+    // is idle but the state may not be static so we need
+    // to use an await to get the current state.
+    let state = worker
+        .state()
+        .await
+        .with_context(|| "failed to deserialize state")?;
+
+    assert_eq!(
+        state,
+        Counters(BTreeMap::from(
+            [("a".to_string(), 1), ("b".to_string(), 2),]
+        ))
+    );
+
+    println!("The system state is now {:?}", state);
+    Ok(())
+}

examples/readme/src/main.rs

@@ -0,0 +1,20 @@
+use anyhow::Result;
+use assert_cmd::prelude::*; // Add methods on commands
+use predicates::prelude::*;
+use std::process::{Command, Output}; // Run programs
+
+#[test]
+fn run_example() -> Result<()> {
+    let mut cmd = Command::cargo_bin(env!("CARGO_PKG_NAME"))?;
+    let assert = cmd.env("RUST_LOG", "debug").assert();
+
+    let Output { stdout, stderr, .. } = cmd.unwrap();
+    println!("{}", String::from_utf8_lossy(&stderr));
+    println!("{}", String::from_utf8_lossy(&stdout));
+
+    assert.success().stdout(predicate::str::diff(
+        "The system state is now Counters({\"a\": 1, \"b\": 2})\n",
+    ));
+
+    Ok(())
+}

examples/readme/tests/cli.rs

@@ -9,7 +9,7 @@ export RUST_LOG="${RUST_LOG},bollard=off,hyper=off"
 find ./examples -name 'Cargo.toml' | while read -r file; do
     dir=$(dirname "$file")
     echo "Running tests in $dir"
-    (cd "$dir" && cargo test -- --test-threads 1) || {
+    (cd "$dir" && cargo test -- --nocapture --test-threads 1) || {
         echo "Tests failed in $dir"
         exit 1
     }

examples/run-tests.sh

@@ -1,10 +1,13 @@
+//! Global error type definitions
+
 use std::ops::Deref;
 
 use thiserror::Error;
 
 #[derive(Debug, Error)]
 #[error("argument extraction failed: {0:?}")]
-/// Arguments to a task could not be extracted.
+/// Job argument extraction failed
+///
 /// This is likely to be an error with the task definition.
 pub struct ExtractionError(#[from] anyhow::Error);
 
@@ -15,16 +18,17 @@ pub struct SerializationError(#[from] serde_json::Error);
 
 #[derive(Debug, Error)]
 #[error("internal error, this may be a bug: {0:?}")]
-/// Some unexpected error happened during the worker operation, either
-/// planning or runtime. These errors should not happen, unless
-/// there is a bug in the implementation.
+/// Some unexpected error happened during the worker operation
+///
+/// These errors should not happen, unless there is a bug in the implementation.
 pub struct InternalError(#[from] anyhow::Error);
 
 #[derive(Debug, Error)]
 #[error("method expansion failed: {0:?}")]
-/// An error happened while trying to expand the method
-/// task into its sub-tasks. This is likely an issue with the
-/// method definition
+/// An error happened while trying to expand the method into sub-tasks
+///
+/// This is likely an issue with the method definition or indicates a that a job on which the
+/// method depends has not been registered with the Worker.
 pub struct MethodError(Box<dyn std::error::Error + Send + Sync>);
 
 impl MethodError {
@@ -36,6 +40,7 @@ impl MethodError {
 #[derive(Debug, Error)]
 #[error(transparent)]
 /// An error happened while executing the task within the workflow.
+///
 /// These errors only happen at runtime, never at the planning stage
 /// of the worker.
 pub struct IOError(Box<dyn std::error::Error + Send + Sync>);