Make operator process one packet at a time asynchronously, use async mutex, drop lock before multiplexing, remove unneeded poison error handling, split model on core side, and add context to load_model error case.

guzman-raphael · guzman-raphael · commit b00b2f581558 · 2025-08-12T00:30:24.000Z
diff --git a/Cargo.toml b/Cargo.toml
@@ -160,6 +160,7 @@ std_instead_of_core = { level = "allow", priority = 127 }              # we shou
 string_add = { level = "allow", priority = 127 }                       # simple concat ok
 string_lit_chars_any = { level = "allow", priority = 127 }             # favor readability until a perf case comes up
 use_debug = { level = "warn", priority = 127 }                         # debug print
+wildcard_enum_match_arm = { level = "allow", priority = 127 }          # allow wildcard match arm in enums
 
 # temporary
 single_call_fn = { level = "allow", priority = 127 }            # remove once more models need pointer serializers/deserializers
diff --git a/cspell.json b/cspell.json
@@ -79,7 +79,8 @@
         "getrandom",
         "wasi",
         "patchelf",
-        "itertools"
+        "itertools",
+        "colinianking"
     ],
     "useGitignore": false,
     "ignorePaths": [
diff --git a/src/core/error.rs b/src/core/error.rs
@@ -6,10 +6,8 @@ use serde_json;
 use serde_yaml;
 use std::{
     backtrace::{Backtrace, BacktraceStatus},
-    error::Error,
     fmt::{self, Formatter},
     io, path,
-    sync::PoisonError,
 };
 use tokio::task;
 
@@ -73,16 +71,6 @@ impl From<path::StripPrefixError> for OrcaError {
         }
     }
 }
-impl<T> From<PoisonError<T>> for OrcaError {
-    fn from(_: PoisonError<T>) -> Self {
-        Self {
-            kind: Kind::PoisonError {
-                source: Box::<dyn Error + Send + Sync>::from("PoisonError"),
-                backtrace: Some(Backtrace::capture()),
-            },
-        }
-    }
-}
 impl From<serde_json::Error> for OrcaError {
     fn from(error: serde_json::Error) -> Self {
         Self {
@@ -142,7 +130,6 @@ impl fmt::Debug for OrcaError {
             | Kind::GlobPatternError { backtrace, .. }
             | Kind::IoError { backtrace, .. }
             | Kind::PathPrefixError { backtrace, .. }
-            | Kind::PoisonError { backtrace, .. }
             | Kind::SerdeJsonError { backtrace, .. }
             | Kind::SerdeYamlError { backtrace, .. }
             | Kind::TokioTaskJoinError { backtrace, .. } => {
diff --git a/src/core/model/mod.rs b/src/core/model/mod.rs
@@ -1,19 +1,12 @@
-use crate::{
-    core::util::get_type_name,
-    uniffi::{
-        error::Result,
-        model::pod::{Pod, PodJob},
-    },
-};
+use crate::{core::util::get_type_name, uniffi::error::Result};
 use heck::ToSnakeCase as _;
 use indexmap::IndexMap;
-use serde::{Deserialize as _, Deserializer, Serialize, Serializer};
+use serde::{Serialize, Serializer};
 use serde_yaml::{self, Value};
 use std::{
     collections::{BTreeMap, HashMap},
     hash::BuildHasher,
     result,
-    sync::Arc,
 };
 /// Converts a model instance into a consistent yaml.
 ///
@@ -64,48 +57,4 @@ where
     sorted.serialize(serializer)
 }
 
-#[expect(clippy::expect_used, reason = "Serde requires this signature.")]
-pub fn deserialize_pod<'de, D>(deserializer: D) -> result::Result<Arc<Pod>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    let value = Value::deserialize(deserializer)?;
-    (value).as_str().map_or_else(
-        || {
-            Ok(serde_yaml::from_value(value.clone())
-                .expect("Failed to convert from serde value to specific type."))
-        },
-        |hash| {
-            Ok({
-                Pod {
-                    hash: hash.to_owned(),
-                    ..Pod::default()
-                }
-                .into()
-            })
-        },
-    )
-}
-
-#[expect(clippy::expect_used, reason = "Serde requires this signature.")]
-pub fn deserialize_pod_job<'de, D>(deserializer: D) -> result::Result<Arc<PodJob>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    let value = Value::deserialize(deserializer)?;
-    (value).as_str().map_or_else(
-        || {
-            Ok(serde_yaml::from_value(value.clone())
-                .expect("Failed to convert from serde value to specific type."))
-        },
-        |hash| {
-            Ok({
-                PodJob {
-                    hash: hash.to_owned(),
-                    ..PodJob::default()
-                }
-                .into()
-            })
-        },
-    )
-}
+pub mod pod;
diff --git a/src/core/model/pod.rs b/src/core/model/pod.rs
@@ -0,0 +1,50 @@
+use crate::uniffi::model::pod::{Pod, PodJob};
+use serde::{Deserialize as _, Deserializer};
+use serde_yaml::{self, Value};
+use std::{result, sync::Arc};
+
+#[expect(clippy::expect_used, reason = "Serde requires this signature.")]
+pub fn deserialize_pod<'de, D>(deserializer: D) -> result::Result<Arc<Pod>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    (value).as_str().map_or_else(
+        || {
+            Ok(serde_yaml::from_value(value.clone())
+                .expect("Failed to convert from serde value to specific type."))
+        },
+        |hash| {
+            Ok({
+                Pod {
+                    hash: hash.to_owned(),
+                    ..Pod::default()
+                }
+                .into()
+            })
+        },
+    )
+}
+
+#[expect(clippy::expect_used, reason = "Serde requires this signature.")]
+pub fn deserialize_pod_job<'de, D>(deserializer: D) -> result::Result<Arc<PodJob>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    (value).as_str().map_or_else(
+        || {
+            Ok(serde_yaml::from_value(value.clone())
+                .expect("Failed to convert from serde value to specific type."))
+        },
+        |hash| {
+            Ok({
+                PodJob {
+                    hash: hash.to_owned(),
+                    ..PodJob::default()
+                }
+                .into()
+            })
+        },
+    )
+}
diff --git a/src/core/operator.rs b/src/core/operator.rs
@@ -1,61 +1,63 @@
 use crate::uniffi::{error::Result, model::packet::Packet};
+use async_trait;
 use itertools::Itertools as _;
-use std::{
-    clone::Clone,
-    collections::HashMap,
-    iter::IntoIterator,
-    sync::{Arc, Mutex},
-};
+use std::{clone::Clone, collections::HashMap, iter::IntoIterator, sync::Arc};
+use tokio::sync::Mutex;
 
+#[async_trait::async_trait]
 pub trait Operator {
-    fn next(&self, packets: Vec<(String, Packet)>) -> Result<Vec<Packet>>;
+    async fn next(&self, stream_name: String, packet: Packet) -> Result<Vec<Packet>>;
 }
 
 pub struct JoinOperator {
     parent_count: usize,
-    received_streams: Arc<Mutex<HashMap<String, Vec<Packet>>>>,
+    received_packets: Arc<Mutex<HashMap<String, Vec<Packet>>>>,
 }
 
 impl JoinOperator {
     pub fn new(parent_count: usize) -> Self {
         Self {
             parent_count,
-            received_streams: Arc::new(Mutex::new(HashMap::new())),
+            received_packets: Arc::new(Mutex::new(HashMap::new())),
         }
     }
 }
 
-#[expect(clippy::excessive_nesting, reason = "Nesting manageable.")]
+#[async_trait::async_trait]
 impl Operator for JoinOperator {
-    fn next(&self, packets: Vec<(String, Packet)>) -> Result<Vec<Packet>> {
-        let mut next_packets = vec![];
-        for (stream, packet) in &packets {
-            let mut received_streams = self.received_streams.lock()?;
-            if self.parent_count - usize::from(!received_streams.contains_key(stream))
-                == received_streams.len()
+    async fn next(&self, stream_name: String, packet: Packet) -> Result<Vec<Packet>> {
+        let mut received_packets = self.received_packets.lock().await;
+        received_packets
+            .entry(stream_name.clone())
+            .or_default()
+            .push(packet.clone());
+        Ok(
+            if self.parent_count - usize::from(!received_packets.contains_key(&stream_name))
+                == received_packets.len()
             {
-                let packets_to_multiplex = received_streams
+                let packets_to_multiplex = received_packets
                     .iter()
                     .filter_map(|(parent_stream, parent_packets)| {
-                        (parent_stream != stream).then_some(parent_packets.clone())
+                        (parent_stream != &stream_name).then_some(parent_packets.clone())
                     })
-                    .chain(vec![vec![packet.clone()]].into_iter());
-                let current_packets = packets_to_multiplex.multi_cartesian_product().map(
-                    |packet_combinations_to_merge| {
+                    .chain(vec![vec![packet.clone()]].into_iter())
+                    .collect::<Vec<_>>();
+                drop(received_packets);
+
+                packets_to_multiplex
+                    .into_iter()
+                    .multi_cartesian_product()
+                    .map(|packet_combinations_to_merge| {
                         packet_combinations_to_merge
                             .into_iter()
                             .flat_map(IntoIterator::into_iter)
                             .collect::<HashMap<_, _>>()
-                    },
-                );
-                next_packets.extend(current_packets);
-            }
-            received_streams
-                .entry(stream.clone())
-                .or_default()
-                .push(packet.clone());
-        }
-        Ok(next_packets)
+                    })
+                    .collect()
+            } else {
+                vec![]
+            },
+        )
     }
 }
 
@@ -69,23 +71,21 @@ impl MapOperator {
     }
 }
 
+#[async_trait::async_trait]
 impl Operator for MapOperator {
-    fn next(&self, packets: Vec<(String, Packet)>) -> Result<Vec<Packet>> {
-        Ok(packets
-            .iter()
-            .map(|(_, packet)| {
-                packet
-                    .iter()
-                    .map(|(packet_key, path_set)| {
-                        (
-                            self.map
-                                .get(packet_key)
-                                .map_or_else(|| packet_key.clone(), Clone::clone),
-                            path_set.clone(),
-                        )
-                    })
-                    .collect()
-            })
-            .collect())
+    async fn next(&self, _: String, packet: Packet) -> Result<Vec<Packet>> {
+        Ok(vec![
+            packet
+                .iter()
+                .map(|(packet_key, path_set)| {
+                    (
+                        self.map
+                            .get(packet_key)
+                            .map_or_else(|| packet_key.clone(), Clone::clone),
+                        path_set.clone(),
+                    )
+                })
+                .collect(),
+        ])
     }
 }
diff --git a/src/core/store/filestore.rs b/src/core/store/filestore.rs
@@ -16,7 +16,7 @@ use heck::ToSnakeCase as _;
 use regex::Regex;
 use serde::{Serialize, de::DeserializeOwned};
 use serde_yaml;
-use snafu::OptionExt as _;
+use snafu::{OptionExt as _, ResultExt as _};
 use std::{
     fmt, fs,
     path::{Path, PathBuf},
@@ -189,15 +189,17 @@ impl LocalFileStore {
         model_id: &ModelID,
     ) -> Result<(T, Option<Annotation>, String)> {
         match model_id {
-            ModelID::Hash(hash) => Ok((
-                serde_yaml::from_str(&fs::read_to_string(self.make_path(
-                    &T::default(),
-                    hash,
-                    Self::SPEC_RELPATH,
-                ))?)?,
-                None,
-                hash.to_owned(),
-            )),
+            ModelID::Hash(hash) => {
+                let path = self.make_path(&T::default(), hash, Self::SPEC_RELPATH);
+                Ok((
+                    serde_yaml::from_str(
+                        &fs::read_to_string(path.clone())
+                            .context(selector::InvalidFilepath { path })?,
+                    )?,
+                    None,
+                    hash.to_owned(),
+                ))
+            }
             ModelID::Annotation(name, version) => {
                 let hash = self.lookup_hash(&T::default(), name, version)?;
                 Ok((
diff --git a/src/uniffi/error.rs b/src/uniffi/error.rs
@@ -78,11 +78,6 @@ pub(crate) enum Kind {
         backtrace: Option<Backtrace>,
     },
     #[snafu(transparent)]
-    PoisonError {
-        source: Box<dyn Error + Send + Sync>,
-        backtrace: Option<Backtrace>,
-    },
-    #[snafu(transparent)]
     SerdeJsonError {
         source: Box<serde_json::Error>,
         backtrace: Option<Backtrace>,
diff --git a/src/uniffi/model/pod.rs b/src/uniffi/model/pod.rs
@@ -2,8 +2,8 @@ use crate::{
     core::{
         crypto::{hash_blob, hash_buffer},
         model::{
-            deserialize_pod, deserialize_pod_job, serialize_hashmap, serialize_hashmap_option,
-            to_yaml,
+            pod::{deserialize_pod, deserialize_pod_job},
+            serialize_hashmap, serialize_hashmap_option, to_yaml,
         },
         util::get,
         validation::validate_packet,
diff --git a/src/uniffi/orchestrator/mod.rs b/src/uniffi/orchestrator/mod.rs
@@ -5,6 +5,7 @@ use crate::uniffi::{
         pod::{PodJob, PodResult},
     },
 };
+use async_trait;
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, fmt, path::PathBuf, sync::Arc};
 use uniffi;
diff --git a/tests/error.rs b/tests/error.rs
diff --git a/tests/operator.rs b/tests/operator.rs
