feat(workflow): Add support for failing workflow execution on certain error types

Author: mjameswh

packages/core-bridge/src/helpers/try_from_js.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashMap, net::SocketAddr, time::Duration};
+use std::{
+    collections::{HashMap, HashSet},
+    net::SocketAddr,
+    time::Duration,
+};
 
 use neon::{
     handle::Handle,
@@ -163,6 +167,24 @@ impl<T: TryFromJs> TryFromJs for Vec<T> {
     }
 }
 
+#[allow(clippy::implicit_hasher)]
+impl<T: TryFromJs + std::hash::Hash + Eq> TryFromJs for HashSet<T> {
+    fn try_from_js<'cx, 'b>(
+        cx: &mut impl Context<'cx>,
+        js_value: Handle<'b, JsValue>,
+    ) -> BridgeResult<Self> {
+        let array = js_value.downcast::<JsArray, _>(cx)?;
+        let len = array.len(cx);
+        let mut result = Self::with_capacity(len as usize);
+
+        for i in 0..len {
+            let value = array.get_value(cx, i)?;
+            result.insert(T::try_from_js(cx, value)?);
+        }
+        Ok(result)
+    }
+}
+
 #[allow(clippy::implicit_hasher)]
 impl<T: TryFromJs> TryFromJs for HashMap<String, T> {
     fn try_from_js<'cx, 'b>(

packages/core-bridge/src/worker.rs

@@ -469,29 +469,24 @@ impl MutableFinalize for HistoryForReplayTunnelHandle {}
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
 mod config {
-    use std::collections::HashSet;
+    use super::custom_slot_supplier::CustomSlotSupplierOptions;
+    use crate::helpers::TryIntoJs;
+    use bridge_macros::TryFromJs;
+    use neon::context::Context;
+    use neon::object::Object;
+    use neon::prelude::JsResult;
+    use neon::types::JsObject;
+    use std::collections::{HashMap, HashSet};
     use std::{sync::Arc, time::Duration};
     use temporalio_common::protos::temporal::api::enums::v1::VersioningBehavior as CoreVersioningBehavior;
     use temporalio_common::protos::temporal::api::worker::v1::PluginInfo;
+    use temporalio_common::worker::WorkerVersioningStrategy;
     use temporalio_common::worker::{
         ActivitySlotKind, LocalActivitySlotKind, NexusSlotKind,
         PollerBehavior as CorePollerBehavior, SlotKind, WorkerConfig, WorkerConfigBuilder,
         WorkerConfigBuilderError, WorkerDeploymentOptions as CoreWorkerDeploymentOptions,
         WorkerDeploymentVersion as CoreWorkerDeploymentVersion, WorkflowSlotKind,
     };
-    use temporalio_sdk_core::{
-        ResourceBasedSlotsOptions, ResourceBasedSlotsOptionsBuilder, ResourceSlotOptions,
-        SlotSupplierOptions as CoreSlotSupplierOptions, TunerHolder, TunerHolderOptionsBuilder,
-    };
-
-    use super::custom_slot_supplier::CustomSlotSupplierOptions;
-    use crate::helpers::TryIntoJs;
-    use bridge_macros::TryFromJs;
-    use neon::context::Context;
-    use neon::object::Object;
-    use neon::prelude::JsResult;
-    use neon::types::JsObject;
-    use temporalio_common::worker::WorkerVersioningStrategy;
 
     #[derive(TryFromJs)]
     pub struct BridgeWorkerOptions {
@@ -515,6 +510,8 @@ mod config {
         max_task_queue_activities_per_second: Option<f64>,
         shutdown_grace_time: Option<Duration>,
         plugins: Vec<String>,
+        workflow_failure_errors: HashSet<WorkflowErrorType>,
+        workflow_types_to_failure_errors: HashMap<String, HashSet<WorkflowErrorType>>,
     }
 
     #[derive(TryFromJs)]
@@ -611,6 +608,10 @@ mod config {
                         })
                         .collect::<HashSet<_>>(),
                 )
+                .workflow_failure_errors(into_core_workflow_error_set(self.workflow_failure_errors))
+                .workflow_types_to_failure_errors(into_core_workflow_error_map_of_sets(
+                    self.workflow_types_to_failure_errors,
+                ))
                 .build()
         }
     }
@@ -682,6 +683,33 @@ mod config {
         }
     }
 
+    #[derive(TryFromJs, Hash, Eq, PartialEq)]
+    pub enum WorkflowErrorType {
+        Nondeterminism,
+    }
+
+    impl From<WorkflowErrorType> for CoreWorkflowErrorType {
+        fn from(val: WorkflowErrorType) -> Self {
+            match val {
+                WorkflowErrorType::Nondeterminism => Self::Nondeterminism,
+            }
+        }
+    }
+
+    fn into_core_workflow_error_set(
+        val: HashSet<WorkflowErrorType>,
+    ) -> HashSet<CoreWorkflowErrorType> {
+        val.into_iter().map(Into::into).collect()
+    }
+
+    fn into_core_workflow_error_map_of_sets(
+        val: HashMap<String, HashSet<WorkflowErrorType>>,
+    ) -> HashMap<String, HashSet<CoreWorkflowErrorType>> {
+        val.into_iter()
+            .map(|(k, v)| (k, into_core_workflow_error_set(v)))
+            .collect()
+    }
+
     #[derive(TryFromJs)]
     #[allow(clippy::struct_field_names)]
     pub(super) struct WorkerTuner {

packages/core-bridge/ts/native.ts

@@ -228,6 +228,8 @@ export interface WorkerOptions {
   maxActivitiesPerSecond: Option<number>;
   shutdownGraceTime: number;
   plugins: string[];
+  workflowFailureErrors: WorkflowErrorType[];
+  workflowTypesToFailureErrors: Record<string, WorkflowErrorType[]>;
 }
 
 export type PollerBehavior =
@@ -255,6 +257,8 @@ export type WorkerDeploymentVersion = {
 
 export type VersioningBehavior = { type: 'pinned' } | { type: 'auto-upgrade' };
 
+export type WorkflowErrorType = { type: 'nondeterminism' };
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 // Worker Tuner
 ////////////////////////////////////////////////////////////////////////////////////////////////////

packages/worker/src/worker-options.ts

@@ -532,6 +532,23 @@ export interface WorkerOptions {
    */
   sinks?: InjectedSinks<any>;
 
+  /**
+   * The types of exceptions that, if a Workflow-thrown error extends, will cause the Workflow
+   * Execution or the Update to fail instead of suspending the Workflow via task failure.
+   *
+   * This property expects a record of Workflow-type names to the list of error types that will
+   * cause that type of Workflow to fail. Uses the `'*'` key to specify a list of error types that
+   * applies to all Workflow types.
+   *
+   * If either list of error types includes `NondeterminismError`, then non-determinism errors
+   * will cause the Workflow Excution to fail. If the list of error types includes `Error`, it
+   * effectively will fail a workflow/update in all user exception cases, including non-determinism
+   * errors.
+   *
+   * @experimental
+   */
+  workflowTypesToFailureErrors?: Record<'*' | string, (string | 'NondeterminismError' | 'Error')[]>;
+
   /**
    * @deprecated SDK tracing is no longer supported. This option is ignored.
    */
@@ -1080,6 +1097,25 @@ function nexusServiceRegistryFromOptions(opts: WorkerOptions): nexus.ServiceRegi
 export function toNativeWorkerOptions(opts: CompiledWorkerOptionsWithBuildId): native.WorkerOptions {
   const enableWorkflows = opts.workflowBundle !== undefined || opts.workflowsPath !== undefined;
   const enableLocalActivities = enableWorkflows && opts.activities.size > 0;
+
+  const workflowFailureErrors: native.WorkflowErrorType[] = [];
+  const workflowTypesToFailureErrors: Record<string, native.WorkflowErrorType[]> = {};
+
+  for (const [k, v] of Object.entries(opts.workflowTypesToFailureErrors ?? {})) {
+    const errorTypes: native.WorkflowErrorType[] = [];
+
+    // Core only cares about Non-Determinism Error; other error types are handled by lang side
+    if (v.includes('NondeterminismError') || v.includes('Error')) {
+      errorTypes.push({ type: 'nondeterminism' });
+    }
+
+    if (k === '*') {
+      workflowFailureErrors.push(...errorTypes);
+    } else {
+      workflowTypesToFailureErrors[k] = errorTypes;
+    }
+  }
+
   return {
     identity: opts.identity,
     buildId: opts.buildId, // eslint-disable-line deprecation/deprecation
@@ -1106,6 +1142,8 @@ export function toNativeWorkerOptions(opts: CompiledWorkerOptionsWithBuildId): n
     maxActivitiesPerSecond: opts.maxActivitiesPerSecond ?? null,
     shutdownGraceTime: msToNumber(opts.shutdownGraceTime),
     plugins: opts.plugins?.map((p) => p.name) ?? [],
+    workflowFailureErrors,
+    workflowTypesToFailureErrors,
   };
 }