Merge pull request #22 from balena-io-experimental/error-handling

Refactor error handling

Author: flowzone-app[bot]

Cargo.toml

@@ -18,6 +18,7 @@ serde = { version = "1.0.197" }
 serde_json = "1.0.120"
 thiserror = "2"
 tokio = { version = "1.43.0", default-features = false, features = [
+  "rt",
   "sync",
   "macros",
 ] }

examples/composer/src/lib.rs

@@ -776,7 +776,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target.clone()).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetStateReached);
+        assert_eq!(worker.status(), &SeekStatus::TargetReached);
 
         // The alpine image must exist now
         let docker = Docker::connect_with_defaults().unwrap();
@@ -934,7 +934,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetStateReached);
+        assert_eq!(worker.status(), &SeekStatus::TargetReached);
 
         // The alpine image must exist now
         let docker = Docker::connect_with_defaults().unwrap();
@@ -973,7 +973,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetStateReached);
+        assert_eq!(worker.status(), &SeekStatus::TargetReached);
 
         // The container ids should match
         let container = docker
@@ -1000,7 +1000,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetStateReached);
+        assert_eq!(worker.status(), &SeekStatus::TargetReached);
 
         // The container ids should match
         let container = docker
@@ -1021,7 +1021,7 @@ mod tests {
 
         // Seeking the target must succeed
         let worker = worker.seek_target(target).await.unwrap();
-        assert_eq!(worker.status(), &SeekStatus::TargetStateReached);
+        assert_eq!(worker.status(), &SeekStatus::TargetReached);
 
         // The container should no longer exist
         let container = docker

src/errors.rs

@@ -0,0 +1,55 @@
+use std::ops::Deref;
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+#[error("argument extraction failed: {0:?}")]
+/// Arguments to a task could not be extracted.
+/// This is likely to be an error with the task definition.
+pub struct ExtractionError(#[from] anyhow::Error);
+
+#[derive(Debug, Error)]
+#[error("serialization error: {0:?}")]
+/// An error happened while serializing or deserializing an input type
+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.
+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
+pub struct MethodError(Box<dyn std::error::Error + Send + Sync>);
+
+impl MethodError {
+    pub fn new<E: std::error::Error + Send + Sync + 'static>(err: E) -> Self {
+        Self(Box::new(err))
+    }
+}
+
+#[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>);
+
+impl IOError {
+    pub fn new<E: std::error::Error + Send + Sync + 'static>(err: E) -> Self {
+        Self(Box::new(err))
+    }
+}
+
+impl Deref for IOError {
+    type Target = Box<dyn std::error::Error + Send + Sync>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}

src/extract/args/mod.rs

@@ -2,8 +2,9 @@ use anyhow::Context as AnyhowCtx;
 use serde::de::DeserializeOwned;
 use std::ops::Deref;
 
+use crate::errors::ExtractionError;
 use crate::system::{FromSystem, System};
-use crate::task::{Context, FromContext, InputError};
+use crate::task::{Context, FromContext};
 
 mod de;
 mod error;
@@ -12,7 +13,7 @@ mod error;
 pub struct Args<T>(pub T);
 
 impl<T: DeserializeOwned + Send> FromContext for Args<T> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_context(context: &Context) -> Result<Self, Self::Error> {
         let args = &context.args;
@@ -28,7 +29,7 @@ impl<T: DeserializeOwned + Send> FromContext for Args<T> {
 }
 
 impl<T: DeserializeOwned + Send> FromSystem for Args<T> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(_: &System, context: &Context) -> Result<Self, Self::Error> {
         Self::from_context(context)

src/extract/path.rs

@@ -1,12 +1,13 @@
 use std::fmt::{Display, Formatter};
 
+use crate::errors::ExtractionError;
 use crate::system::{FromSystem, System};
-use crate::task::{Context, FromContext, InputError};
+use crate::task::{Context, FromContext};
 
 pub struct Path(pub String);
 
 impl FromContext for Path {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_context(context: &Context) -> Result<Self, Self::Error> {
         let path = context.path.to_str();
@@ -22,7 +23,7 @@ impl Display for Path {
 }
 
 impl FromSystem for Path {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(_: &System, context: &Context) -> Result<Self, Self::Error> {
         Self::from_context(context)

src/extract/res.rs

@@ -2,13 +2,14 @@ use anyhow::Context as AnyhowCxt;
 use std::ops::Deref;
 use std::sync::Arc;
 
+use crate::errors::ExtractionError;
 use crate::system::{FromSystem, System};
-use crate::task::{Context, InputError};
+use crate::task::Context;
 
 pub struct Res<R>(Arc<R>);
 
 impl<R: Send + Sync + 'static> FromSystem for Res<R> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(system: &System, _: &Context) -> Result<Self, Self::Error> {
         let arc = system
@@ -19,7 +20,7 @@ impl<R: Send + Sync + 'static> FromSystem for Res<R> {
                     std::any::type_name::<R>()
                 )
             })
-            .map_err(InputError::from)?;
+            .map_err(ExtractionError::from)?;
         Ok(Res(arc))
     }
 }

src/extract/system.rs

@@ -1,15 +1,16 @@
 use anyhow::Context as AnyhowCtx;
 use serde::de::DeserializeOwned;
 
+use crate::errors::ExtractionError;
 use crate::system::{FromSystem, System as SystemState};
-use crate::task::{Context, InputError};
+use crate::task::Context;
 
 /// Extracts the root of the system state
 #[derive(Debug, Clone)]
 pub struct System<S>(pub S);
 
 impl<S: DeserializeOwned> FromSystem for System<S> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(system: &SystemState, _: &Context) -> Result<Self, Self::Error> {
         // This will fail if the value cannot be deserialized into the target type

src/extract/target.rs

@@ -2,13 +2,14 @@ use anyhow::Context as AnyhowCxt;
 use serde::de::DeserializeOwned;
 use std::ops::Deref;
 
+use crate::errors::ExtractionError;
 use crate::system::{FromSystem, System};
-use crate::task::{Context, FromContext, InputError};
+use crate::task::{Context, FromContext};
 
 pub struct Target<T>(pub T);
 
 impl<T: DeserializeOwned> FromContext for Target<T> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_context(context: &Context) -> Result<Self, Self::Error> {
         let value = &context.target;
@@ -26,7 +27,7 @@ impl<T: DeserializeOwned> FromContext for Target<T> {
 }
 
 impl<T: DeserializeOwned> FromSystem for Target<T> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(_: &System, context: &Context) -> Result<Self, Self::Error> {
         Self::from_context(context)

src/extract/view.rs

@@ -10,9 +10,10 @@ use serde::Serialize;
 use serde_json::Value;
 use std::ops::{Deref, DerefMut};
 
+use crate::errors::ExtractionError;
 use crate::path::Path;
 use crate::system::{FromSystem, System};
-use crate::task::{Context, Effect, Error, InputError, IntoResult, UnexpectedError};
+use crate::task::{Context, Effect, Error, IntoResult};
 
 /// Extracts a pointer to a sub-element of the global state indicated
 /// by the path.
@@ -75,7 +76,7 @@ impl<T> Pointer<T> {
 }
 
 impl<T: DeserializeOwned> FromSystem for Pointer<T> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(system: &System, context: &Context) -> Result<Self, Self::Error> {
         let json_ptr = context.path.as_ref();
@@ -180,9 +181,10 @@ impl<T: Serialize> IntoResult<Patch> for Pointer<T> {
     fn into_result(self) -> Result<Patch, Error> {
         let before = self.initial;
         let after = if let Some(state) = self.state {
-            serde_json::to_value(state)
-                .with_context(|| "Failed to serialize pointer value")
-                .map_err(UnexpectedError::from)?
+            // This should not happen unless there is a bug (hopefully).
+            // if this happens during worker operation, it will be catched
+            // as a panic in the task
+            serde_json::to_value(state).expect("failed to serialize pointer value")
         } else {
             Value::Null
         };
@@ -210,14 +212,6 @@ impl<T, E> From<Pointer<T>> for Effect<Pointer<T>, E> {
     }
 }
 
-// Allow tasks to return a pointer
-// This converts the pointer into a pure effect
-impl<T: Serialize> From<Pointer<T>> for Effect<Patch, Error> {
-    fn from(ptr: Pointer<T>) -> Effect<Patch, Error> {
-        Effect::from_result(ptr.into_result())
-    }
-}
-
 /// Extracts a sub-element of a state S as indicated by
 /// a path.
 ///
@@ -227,7 +221,7 @@ impl<T: Serialize> From<Pointer<T>> for Effect<Patch, Error> {
 pub struct View<T>(Pointer<T>);
 
 impl<T: DeserializeOwned> FromSystem for View<T> {
-    type Error = InputError;
+    type Error = ExtractionError;
 
     fn from_system(system: &System, context: &Context) -> Result<Self, Self::Error> {
         let pointer = Pointer::<T>::from_system(system, context)?;
@@ -269,14 +263,6 @@ impl<T, E> From<View<T>> for Effect<View<T>, E> {
     }
 }
 
-// Allow tasks to return a view
-// This converts the view into a pure effect
-impl<T: Serialize> From<View<T>> for Effect<Patch, Error> {
-    fn from(view: View<T>) -> Effect<Patch, Error> {
-        Effect::from_result(view.into_result())
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

src/lib.rs

@@ -2,6 +2,7 @@ mod path;
 mod planner;
 mod system;
 
+pub mod errors;
 pub mod extract;
 pub mod task;
 pub mod worker;