Enable access to the parent step's output value.

Author: jakubadamw

Cargo.lock

@@ -20,6 +20,7 @@ thiserror = "1"
 tokio = { version = "1", features = ["fs", "macros", "rt-multi-thread", "sync"] }
 tonic = { version = "0.12", features = ["tls", "tls-native-roots"] }
 tracing = "0.1"
+ustr = { version = "1", features = ["serde"] }
 
 [build-dependencies]
 tonic-build = "0.12"

Cargo.toml

@@ -0,0 +1,57 @@
+use hatchet_sdk::{Client, Context, StepBuilder, WorkflowBuilder};
+
+#[derive(serde::Serialize, serde::Deserialize)]
+struct HelloOutput {
+    text: String,
+}
+
+async fn execute_hello(_context: Context, _: serde_json::Value) -> anyhow::Result<HelloOutput> {
+    Ok(HelloOutput {
+        text: "Hello".to_owned(),
+    })
+}
+
+async fn execute_world(
+    mut context: Context,
+    _: serde_json::Value,
+) -> anyhow::Result<serde_json::Value> {
+    let hello_result: HelloOutput = context.pop_parent_output("hello");
+    Ok(serde_json::json!({
+        "text": format!("{} World!", hello_result.text)
+    }))
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    dotenv::dotenv().ok();
+    tracing_subscriber::fmt()
+        .with_target(false)
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::from_default_env()
+                .add_directive("hatchet_sdk=debug".parse()?),
+        )
+        .init();
+
+    let client = Client::new()?;
+    let mut worker = client.worker("example_spawn_workflow").build();
+    worker.register_workflow(
+        WorkflowBuilder::default()
+            .name("hello-world")
+            .step(
+                StepBuilder::default()
+                    .name("hello")
+                    .function(&execute_hello)
+                    .build()?,
+            )
+            .step(
+                StepBuilder::default()
+                    .name("world")
+                    .function(&execute_world)
+                    .parent("hello")
+                    .build()?,
+            )
+            .build()?,
+    );
+    worker.start().await?;
+    Ok(())
+}

examples/multiple_steps.rs

@@ -0,0 +1,55 @@
+use hatchet_sdk::{Client, Context, StepBuilder, WorkflowBuilder};
+
+#[derive(serde::Serialize, serde::Deserialize)]
+struct HelloOutput {
+    text: String,
+}
+
+async fn execute_hello(_context: Context, _: serde_json::Value) -> anyhow::Result<HelloOutput> {
+    Ok(HelloOutput {
+        text: "Hello".to_owned(),
+    })
+}
+
+async fn execute_panic(
+    mut context: Context,
+    _: serde_json::Value,
+) -> anyhow::Result<serde_json::Value> {
+    let hello_result: HelloOutput = context.pop_parent_output("hello");
+    panic!("Panic {}", hello_result.text);
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    dotenv::dotenv().ok();
+    tracing_subscriber::fmt()
+        .with_target(false)
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::from_default_env()
+                .add_directive("hatchet_sdk=debug".parse()?),
+        )
+        .init();
+
+    let client = Client::new()?;
+    let mut worker = client.worker("example_spawn_workflow").build();
+    worker.register_workflow(
+        WorkflowBuilder::default()
+            .name("hello-panic")
+            .step(
+                StepBuilder::default()
+                    .name("hello")
+                    .function(&execute_hello)
+                    .build()?,
+            )
+            .step(
+                StepBuilder::default()
+                    .name("panic")
+                    .function(&execute_panic)
+                    .parent("hello")
+                    .build()?,
+            )
+            .build()?,
+    );
+    worker.start().await?;
+    Ok(())
+}

examples/panicking_step.rs

@@ -6,6 +6,7 @@ use std::{
 
 use futures_util::lock::Mutex;
 use tracing::info;
+use ustr::Ustr;
 
 use crate::worker::{grpc, ServiceWithAuthorization};
 
@@ -24,10 +25,12 @@ pub struct Context {
         u16,
     )>,
     data: Arc<DataMap>,
+    parents_outputs: HashMap<Ustr, serde_json::Value>,
 }
 
 impl Context {
     pub(crate) fn new(
+        parents_outputs: HashMap<Ustr, serde_json::Value>,
         workflow_run_id: String,
         workflow_step_run_id: String,
         workflow_service_client: grpc::workflow_service_client::WorkflowServiceClient<
@@ -39,13 +42,23 @@ impl Context {
         data: Arc<DataMap>,
     ) -> Self {
         Self {
+            parents_outputs,
             workflow_run_id,
             workflow_service_client_and_spawn_index: Mutex::new((workflow_service_client, 0)),
             workflow_step_run_id,
             data,
         }
     }
 
+    pub fn pop_parent_output<O: serde::de::DeserializeOwned>(&mut self, step_name: &str) -> O {
+        serde_json::from_value(
+            self.parents_outputs
+                .remove(&step_name.into())
+                .unwrap_or_else(|| panic!("could not find the output for step '{step_name}'")),
+        )
+        .expect("could not deserialize from JSON")
+    }
+
     pub fn datum<D: std::any::Any + Send + Sync>(&self) -> &D {
         let type_id = TypeId::of::<D>();
         self.data
@@ -87,8 +100,10 @@ impl Context {
     }
 }
 
-pub(crate) type StepFunction =
-    dyn Fn(
+pub(crate) type StepFunction = dyn Fn(
         Context,
         serde_json::Value,
-    ) -> futures_util::future::LocalBoxFuture<'static, anyhow::Result<serde_json::Value>>;
+    ) -> std::panic::AssertUnwindSafe<
+        futures_util::future::BoxFuture<'static, anyhow::Result<serde_json::Value>>,
+    > + Send
+    + Sync;