MVP Pipeline Implementation with Utils update

Cargo.toml

@@ -37,6 +37,7 @@ chrono = "0.4.39"
 colored = "2.1.0"
 # derive utilities for new types
 derive_more = { version = "2.0.1", features = ["display"] }
+dot-parser = "0.5.1"
 # chaining async calls and processing stream data in local docker orchestrator
 futures-util = "0.3.31"
 # auto derive getter access methods on structs
@@ -47,8 +48,10 @@ glob = "0.3.1"
 heck = "0.5.0"
 # hashmaps that preserve insertion order
 indexmap = { version = "2.9.0", features = ["serde"] }
+layout-rs = "0.1.3"
 # random name generator
 names = "0.14.0"
+petgraph = "0.8.2"
 # complex pattern matching in strings
 regex = "1.11.0"
 # serialization/deserialization to/from filestore
@@ -109,6 +112,7 @@ impl_trait_in_params = { level = "allow", priority = 127 }             # impl in
 implicit_return = { level = "allow", priority = 127 }                  # missing return ok
 inline_asm_x86_intel_syntax = { level = "allow", priority = 127 }      # intel syntax ok
 integer_division = { level = "allow", priority = 127 }                 # allow discarding remainder
+iter_over_hash_type = { level = "allow", priority = 127 }              # allow iterating over hash types
 little_endian_bytes = { level = "allow", priority = 127 }              # allow to_le_bytes / from_le_bytes
 missing_asserts_for_indexing = { level = "allow", priority = 127 }     # missing assert before indexing ok
 missing_docs_in_private_items = { level = "allow", priority = 127 }    # missing docs on private ok

cspell.json

@@ -73,7 +73,9 @@
         "getset",
         "strsim",
         "getrandom",
-        "wasi"
+        "wasi",
+        "petgraph",
+        "rfind"
     ],
     "useGitignore": false,
     "ignorePaths": [

src/core/error.rs

@@ -90,6 +90,7 @@
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match &self.kind {
             Kind::EmptyResponseWhenLoadingContainerAltImage { backtrace, .. }
+            | Kind::FailedToParseDot { backtrace, .. }
             | Kind::GeneratedNamesOverflow { backtrace, .. }
             | Kind::InvalidFilepath { backtrace, .. }
             | Kind::InvalidPodResultTerminatedDatetime { backtrace, .. }
@@ -99,6 +100,7 @@
             | Kind::NoFileName { backtrace, .. }
             | Kind::NoMatchingPodRun { backtrace, .. }
             | Kind::NoTagFoundInContainerAltImage { backtrace, .. }
+            | Kind::MissingInputSpecKey { backtrace, .. }
             | Kind::BollardError { backtrace, .. }
             | Kind::GlobPatternError { backtrace, .. }
             | Kind::IoError { backtrace, .. }

src/core/util.rs

@@ -1,6 +1,14 @@
-use crate::uniffi::error::{Result, selector};
+use crate::uniffi::{
+    error::{Result, selector},
+    model::PathSet,
+};
 use snafu::OptionExt as _;
-use std::{any::type_name, collections::HashMap, fmt};
+use std::{
+    any::type_name,
+    collections::HashMap,
+    fmt::{self, Debug},
+    hash::Hash,
+};
 
 #[expect(
     clippy::unwrap_used,
@@ -26,8 +34,27 @@ pub fn parse_debug_name<T: fmt::Debug>(instance: &T) -> String {
         .unwrap()
 }
 
-pub fn get<'map, T>(map: &'map HashMap<String, T>, key: &str) -> Result<&'map T> {
-    Ok(map.get(key).context(selector::KeyMissing {
-        key: key.to_owned(),
-    })?)
+pub fn get<'map, K, T>(map: &'map HashMap<K, T>, key: &K) -> Result<&'map T>
+where
+    K: Hash + Eq + ToOwned<Owned = K> + Debug,
+{
+    let temp = map.get(key).context(selector::KeyMissing {
+        key: format!("{key:?}"),
+    })?;
+    Ok(temp)
+}
+
+pub fn find_missing_keys<'a>(
+    input_map: &HashMap<String, PathSet>,
+    keys_to_check: impl Iterator<Item = &'a String>,
+) -> Vec<String> {
+    keys_to_check
+        .filter_map(|key| {
+            if input_map.contains_key(key) {
+                None
+            } else {
+                Some(key.clone())
+            }
+        })
+        .collect()
 }

src/uniffi/error.rs

@@ -15,6 +15,7 @@ use std::{
     result,
 };
 use uniffi;
+
 /// Shorthand for a Result that returns an `OrcaError`.
 pub type Result<T, E = OrcaError> = result::Result<T, E>;
 /// Possible errors you may encounter.
@@ -29,6 +30,12 @@ pub(crate) enum Kind {
         path: PathBuf,
         backtrace: Option<Backtrace>,
     },
+    #[snafu(display("Failed to parse DOT graph: {dot} with reason: {reason}."))]
+    FailedToParseDot {
+        dot: String,
+        reason: String,
+        backtrace: Option<Backtrace>,
+    },
     #[snafu(display("Out of generated random names."))]
     GeneratedNamesOverflow { backtrace: Option<Backtrace> },
     #[snafu(display("{source} ({path:?})."))]
@@ -58,6 +65,7 @@ pub(crate) enum Kind {
     },
     #[snafu(display("No known container names."))]
     NoContainerNames { backtrace: Option<Backtrace> },
+
     #[snafu(display("Missing file or directory name ({path:?})."))]
     NoFileName {
         path: PathBuf,
@@ -73,6 +81,11 @@ pub(crate) enum Kind {
         path: PathBuf,
         backtrace: Option<Backtrace>,
     },
+    #[snafu(display("Input map missing required packet keys: {missing_keys:?}"))]
+    MissingInputSpecKey {
+        missing_keys: Vec<String>,
+        backtrace: Option<Backtrace>,
+    },
     #[snafu(transparent)]
     BollardError {
         source: BollardError,

src/uniffi/mod.rs

@@ -6,3 +6,6 @@ pub mod model;
 pub mod orchestrator;
 /// Data persistence provided by a store backend.
 pub mod store;
+
+/// Pipeline management and execution.
+pub mod pipeline;

src/uniffi/orchestrator/docker.rs

@@ -157,7 +157,7 @@ impl Orchestrator for LocalDockerOrchestrator {
         .await?
         .map(|(assigned_name, run_info)| {
             let pod_job: PodJob =
-                serde_json::from_str(get(&run_info.labels, "org.orcapod.pod_job")?)?;
+                serde_json::from_str(get(&run_info.labels, &"org.orcapod.pod_job".to_owned())?)?;
             Ok(PodRun::new::<Self>(&pod_job, assigned_name))
         })
         .collect()