Added logging to PodResult

.clippy.toml

@@ -1,3 +1,3 @@
-excessive-nesting-threshold = 4
+excessive-nesting-threshold = 6
 too-many-arguments-threshold = 10
 allowed-idents-below-min-chars = ["..", "k", "v", "f", "re", "id", "Ok", "'_"]

.github/workflows/tests.yaml

@@ -17,7 +17,7 @@ jobs:
       - name: Install Rust + components
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: 1.87
+          toolchain: 1.89
           components: rustfmt,clippy
       - name: Install code coverage
         uses: taiki-e/install-action@cargo-llvm-cov

.vscode/settings.json

@@ -9,7 +9,6 @@
     ],
     "files.autoSave": "off",
     "files.insertFinalNewline": true,
-    "gitlens.showWelcomeOnInstall": false,
     "gitlens.showWhatsNewAfterUpgrades": false,
     "lldb.consoleMode": "evaluate",
     "rust-analyzer.cargo.features": [

Cargo.toml

@@ -55,6 +55,7 @@ glob = "0.3.1"
 heck = "0.5.0"
 # convert bytes to hex strings
 hex = "0.4.3"
+hostname = "0.4.1"
 # hashmaps that preserve insertion order
 indexmap = { version = "2.9.0", features = ["serde"] }
 # utilities for iterables e.g. cartesian products

cspell.json

@@ -80,7 +80,9 @@
         "wasi",
         "patchelf",
         "itertools",
-        "colinianking"
+        "colinianking",
+        "itertools",
+        "pathset",
     ],
     "useGitignore": false,
     "ignorePaths": [

src/core/error.rs

@@ -122,9 +122,16 @@ impl fmt::Debug for OrcaError {
         match &self.kind {
             Kind::AgentCommunicationFailure { backtrace, .. }
             | Kind::FailedToStartPod { backtrace, .. }
+            | Kind::FailedToExtractRunInfo { backtrace, .. }
             | Kind::IncompletePacket { backtrace, .. }
             | Kind::InvalidPath { backtrace, .. }
+            | Kind::InvalidIndex { backtrace, .. }
+            | Kind::KeyMissing { backtrace, .. }
             | Kind::MissingInfo { backtrace, .. }
+            | Kind::FailedToGetPodJobOutput { backtrace, .. }
+            | Kind::PodJobProcessingError { backtrace, .. }
+            | Kind::PodJobSubmissionFailed { backtrace, .. }
+            | Kind::UnexpectedPathType { backtrace, .. }
             | Kind::BollardError { backtrace, .. }
             | Kind::ChronoParseError { backtrace, .. }
             | Kind::DOTError { backtrace, .. }

src/core/graph.rs

@@ -1,5 +1,5 @@
 use crate::{
-    core::{pipeline::PipelineNode, util::get},
+    core::{model::pipeline::PipelineNode, util::get},
     uniffi::{error::Result, model::pipeline::Kernel},
 };
 use dot_parser::ast::Graph as DOTGraph;
@@ -25,7 +25,7 @@ pub fn make_graph(
     let graph =
         DiGraph::<DotNodeWeight, DotAttrList>::from_dot_graph(DOTGraph::try_from(input_dot)?).map(
             |_, node| PipelineNode {
-                name: node.id.clone(),
+                id: node.id.clone(),
                 kernel: get(&metadata, &node.id)
                     .unwrap_or_else(|error| panic!("{error}"))
                     .clone(),

src/core/mod.rs

@@ -9,7 +9,6 @@ macro_rules! inner_attr_to_each {
 
 pub(crate) mod error;
 pub(crate) mod graph;
-pub(crate) mod pipeline;
 pub(crate) mod store;
 pub(crate) mod util;
 pub(crate) mod validation;
@@ -20,6 +19,7 @@ inner_attr_to_each! {
     pub(crate) mod model;
     pub(crate) mod operator;
     pub(crate) mod orchestrator;
+    pub(crate) mod pipeline_runner;
 }
 
 #[cfg(feature = "test")]
@@ -35,6 +35,7 @@ inner_attr_to_each! {
     )]
     pub mod crypto;
     pub mod model;
-    pub mod operator;
     pub mod orchestrator;
+    pub mod pipeline_runner;
+    pub mod operator;
 }

src/core/model/mod.rs

@@ -57,4 +57,5 @@ where
     sorted.serialize(serializer)
 }
 
+pub mod pipeline;
 pub mod pod;

src/core/model/pipeline.rs

@@ -0,0 +1,109 @@
+use std::{
+    backtrace::Backtrace,
+    collections::{HashMap, HashSet},
+};
+
+use crate::uniffi::{
+    error::{Kind, OrcaError, Result},
+    model::{
+        packet::PathSet,
+        pipeline::{Kernel, Pipeline, PipelineJob},
+    },
+};
+use itertools::Itertools as _;
+use petgraph::Direction::Incoming;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+pub struct PipelineNode {
+    pub id: String,
+    pub kernel: Kernel,
+}
+
+impl Pipeline {
+    /// Function to get the parents of a node
+    pub(crate) fn get_node_parents(
+        &self,
+        node: &PipelineNode,
+    ) -> impl Iterator<Item = &PipelineNode> {
+        // Find the NodeIndex for the given node_key
+        let node_index = self
+            .graph
+            .node_indices()
+            .find(|&idx| self.graph[idx] == *node);
+        node_index.into_iter().flat_map(move |idx| {
+            self.graph
+                .neighbors_directed(idx, Incoming)
+                .map(move |parent_idx| &self.graph[parent_idx])
+        })
+    }
+
+    /// Return a vec of `node_names` that takes in inputs based on the `input_spec`
+    pub(crate) fn get_input_nodes(&self) -> HashSet<&String> {
+        let mut input_nodes = HashSet::new();
+
+        self.input_spec.iter().for_each(|(_, node_uris)| {
+            for node_uri in node_uris {
+                input_nodes.insert(&node_uri.node_id);
+            }
+        });
+
+        input_nodes
+    }
+}
+
+impl PipelineJob {
+    /// Helpful function to get the input packet for input nodes of the pipeline based on the `pipeline_job` an`pipeline_spec`ec
+    /// # Errors
+    /// Will return `Err` if there is an issue getting the input packet per node.
+    /// # Returns
+    /// A `HashMap` where the key is the node name and the value is a vector of `HashMap<String, PathSet>` representing the input packets for that node.
+    pub fn get_input_packet_per_node(
+        &self,
+    ) -> Result<HashMap<String, Vec<HashMap<String, PathSet>>>> {
+        // For each node in the input specification, we will iterate over its mapping
+        // nodes_input_spec contains <node_id, HashMap<key, PathSet>>
+        let mut nodes_input_spec = HashMap::new();
+        for (input_key, node_uris) in &self.pipeline.input_spec {
+            for node_uri in node_uris {
+                let input_path_sets = self.input_packet.get(input_key).ok_or(OrcaError {
+                    kind: Kind::KeyMissing {
+                        key: input_key.clone(),
+                        backtrace: Some(Backtrace::capture()),
+                    },
+                })?;
+                // There shouldn't be a duplicate key in the input packet as this will be handle by pipeline verify
+                let input_spec = nodes_input_spec
+                    .entry(&node_uri.node_id)
+                    .or_insert_with(HashMap::new);
+                input_spec.insert(&node_uri.key, input_path_sets);
+            }
+        }
+
+        // For each node, compute the cartesian product of the path_sets for each unique combination of keys
+        let node_input_packets = nodes_input_spec
+            .into_iter()
+            .map(|(node_id, input_node_keys)| {
+                // We need to pull them out at the same time to ensure the key order is preserve to match the cartesian product
+                let (keys, values): (Vec<_>, Vec<_>) = input_node_keys.into_iter().unzip();
+
+                // Covert each combo into a packet
+                let packets = values
+                    .into_iter()
+                    .multi_cartesian_product()
+                    .map(|combo| {
+                        keys.iter()
+                            .copied()
+                            .zip(combo)
+                            .map(|(key, pathset)| (key.to_owned(), pathset.to_owned()))
+                            .collect::<HashMap<_, _>>()
+                    })
+                    .collect::<Vec<HashMap<String, PathSet>>>();
+
+                (node_id.to_owned(), packets)
+            })
+            .collect::<HashMap<_, _>>();
+
+        Ok(node_input_packets)
+    }
+}

src/core/operator.rs

@@ -1,12 +1,13 @@
 use crate::uniffi::{error::Result, model::packet::Packet};
 use async_trait;
 use itertools::Itertools as _;
+use serde::{Deserialize, Serialize};
 use std::{clone::Clone, collections::HashMap, iter::IntoIterator, sync::Arc};
 use tokio::sync::Mutex;
 
 #[async_trait::async_trait]
 pub trait Operator {
-    async fn next(&self, stream_name: String, packet: Packet) -> Result<Vec<Packet>>;
+    async fn process_packet(&self, stream_name: String, packet: Packet) -> Result<Vec<Packet>>;
 }
 
 pub struct JoinOperator {
@@ -25,7 +26,7 @@ impl JoinOperator {
 
 #[async_trait::async_trait]
 impl Operator for JoinOperator {
-    async fn next(&self, stream_name: String, packet: Packet) -> Result<Vec<Packet>> {
+    async fn process_packet(&self, stream_name: String, packet: Packet) -> Result<Vec<Packet>> {
         let mut received_packets = self.received_packets.lock().await;
         received_packets
             .entry(stream_name.clone())
@@ -61,8 +62,9 @@ impl Operator for JoinOperator {
     }
 }
 
+#[derive(uniffi::Object, Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
 pub struct MapOperator {
-    map: HashMap<String, String>,
+    pub map: HashMap<String, String>,
 }
 
 impl MapOperator {
@@ -73,7 +75,7 @@ impl MapOperator {
 
 #[async_trait::async_trait]
 impl Operator for MapOperator {
-    async fn next(&self, _: String, packet: Packet) -> Result<Vec<Packet>> {
+    async fn process_packet(&self, _: String, packet: Packet) -> Result<Vec<Packet>> {
         Ok(vec![
             packet
                 .iter()