Enable failing WF on nondeterminism errors (#702)

Author: Sushisource

.github/workflows/per-pr.yml

@@ -13,7 +13,7 @@ concurrency:
 jobs:
   build-and-test:
     name: "Format, docs, and lint"
-    timeout-minutes: 20
+    timeout-minutes: 10
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
@@ -48,6 +48,7 @@ jobs:
 
   test:
     name: Unit Tests
+    timeout-minutes: 10
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
@@ -77,6 +78,7 @@ jobs:
 
   fmt:
     name: Integ tests
+    timeout-minutes: 20
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2

core-api/src/errors.rs

@@ -60,3 +60,11 @@ pub enum CompleteActivityError {
         completion: Option<ActivityExecutionResult>,
     },
 }
+
+/// Errors we can encounter during workflow processing which we may treat as either WFT failures
+/// or whole-workflow failures depending on user preference.
+#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+pub enum WorkflowErrorType {
+    /// A nondeterminism error
+    Nondeterminism,
+}

core-api/src/worker.rs

@@ -1,4 +1,8 @@
-use std::time::Duration;
+use crate::errors::WorkflowErrorType;
+use std::{
+    collections::{HashMap, HashSet},
+    time::Duration,
+};
 
 const MAX_OUTSTANDING_WFT_DEFAULT: usize = 100;
 const MAX_CONCURRENT_WFT_POLLS_DEFAULT: usize = 5;
@@ -124,6 +128,16 @@ pub struct WorkerConfig {
     /// timeout.
     #[builder(default = "Duration::from_secs(5)")]
     pub local_timeout_buffer_for_activities: Duration,
+
+    /// Any error types listed here will cause any workflow being processed by this worker to fail,
+    /// rather than simply failing the workflow task.
+    #[builder(default)]
+    pub workflow_failure_errors: HashSet<WorkflowErrorType>,
+
+    /// Like [WorkerConfig::workflow_failure_errors], but specific to certain workflow types (the
+    /// map key).
+    #[builder(default)]
+    pub workflow_types_to_failure_errors: HashMap<String, HashSet<WorkflowErrorType>>,
 }
 
 impl WorkerConfig {
@@ -136,6 +150,20 @@ impl WorkerConfig {
             .saturating_sub(self.max_nonsticky_polls())
             .max(1)
     }
+    /// Returns true if the configuration specifies we should fail a workflow on a certain error
+    /// type rather than failing the workflow task.
+    pub fn should_fail_workflow(
+        &self,
+        workflow_type: &str,
+        error_type: &WorkflowErrorType,
+    ) -> bool {
+        self.workflow_failure_errors.contains(error_type)
+            || self
+                .workflow_types_to_failure_errors
+                .get(workflow_type)
+                .map(|s| s.contains(error_type))
+                .unwrap_or(false)
+    }
 }
 
 impl WorkerConfigBuilder {

core/src/worker/workflow/machines/workflow_machines.rs

@@ -1293,21 +1293,12 @@ impl WorkflowMachines {
                     self.process_cancellation(CommandID::LocalActivity(attrs.seq))?;
                 }
                 WFCommand::CompleteWorkflow(attrs) => {
-                    if !self.replaying {
-                        self.metrics.wf_completed();
-                    }
                     self.add_terminal_command(complete_workflow(attrs));
                 }
                 WFCommand::FailWorkflow(attrs) => {
-                    if !self.replaying {
-                        self.metrics.wf_failed();
-                    }
                     self.add_terminal_command(fail_workflow(attrs));
                 }
                 WFCommand::ContinueAsNew(attrs) => {
-                    if !self.replaying {
-                        self.metrics.wf_continued_as_new();
-                    }
                     let attrs = self.augment_continue_as_new_with_current_values(attrs);
                     let use_compat = self.determine_use_compatible_flag(
                         attrs.versioning_intent(),
@@ -1316,9 +1307,6 @@ impl WorkflowMachines {
                     self.add_terminal_command(continue_as_new(attrs, use_compat));
                 }
                 WFCommand::CancelWorkflow(attrs) => {
-                    if !self.replaying {
-                        self.metrics.wf_canceled();
-                    }
                     self.add_terminal_command(cancel_workflow(attrs));
                 }
                 WFCommand::SetPatchMarker(attrs) => {

core/src/worker/workflow/managed_run.rs

@@ -23,19 +23,23 @@ use std::{
     mem,
     ops::Add,
     rc::Rc,
-    sync::mpsc::Sender,
+    sync::{mpsc::Sender, Arc},
     time::{Duration, Instant},
 };
+use temporal_sdk_core_api::{errors::WorkflowErrorType, worker::WorkerConfig};
 use temporal_sdk_core_protos::{
     coresdk::{
         workflow_activation::{
             create_evict_activation, query_to_job, remove_from_cache::EvictionReason,
             workflow_activation_job, RemoveFromCache, WorkflowActivation,
         },
-        workflow_commands::QueryResult,
+        workflow_commands::{FailWorkflowExecution, QueryResult},
         workflow_completion,
     },
-    temporal::api::{enums::v1::WorkflowTaskFailedCause, failure::v1::Failure},
+    temporal::api::{
+        command::v1::command::Attributes as CmdAttribs, enums::v1::WorkflowTaskFailedCause,
+        failure::v1::Failure,
+    },
     TaskToken,
 };
 use tokio::sync::oneshot;
@@ -92,6 +96,7 @@ pub(super) struct ManagedRun {
     /// We store the paginator used for our own run's history fetching
     paginator: Option<HistoryPaginator>,
     completion_waiting_on_page_fetch: Option<RunActivationCompletion>,
+    config: Arc<WorkerConfig>,
 }
 impl ManagedRun {
     pub(super) fn new(
@@ -100,6 +105,7 @@ impl ManagedRun {
         local_activity_request_sink: Rc<dyn LocalActivityRequestSink>,
     ) -> (Self, RunUpdateAct) {
         let metrics = basics.metrics.clone();
+        let config = basics.worker_config.clone();
         let wfm = WorkflowManager::new(basics);
         let mut me = Self {
             wfm,
@@ -114,6 +120,7 @@ impl ManagedRun {
             metrics,
             paginator: None,
             completion_waiting_on_page_fetch: None,
+            config,
         };
         let rua = me.incoming_wft(wft);
         (me, rua)
@@ -534,7 +541,6 @@ impl ManagedRun {
             return None;
         };
 
-        self.metrics.wf_task_failed();
         let message = format!("Workflow activation completion failed: {:?}", &failure);
         // We don't want to fail queries that could otherwise be retried
         let is_no_report_query_fail = self.pending_work_is_legacy_query()
@@ -570,12 +576,35 @@ impl ManagedRun {
                 )
             }
         } else if should_report {
-            ActivationCompleteOutcome::ReportWFTFail(FailedActivationWFTReport::Report(
-                tt, cause, failure,
-            ))
+            // Check if we should fail the workflow instead of the WFT because of user's preferences
+            if matches!(cause, WorkflowTaskFailedCause::NonDeterministicError)
+                && self.config.should_fail_workflow(
+                    &self.wfm.machines.workflow_type,
+                    &WorkflowErrorType::Nondeterminism,
+                )
+            {
+                warn!(failure=?failure, "Failing workflow due to nondeterminism error");
+                return self
+                    .successful_completion(
+                        vec![WFCommand::FailWorkflow(FailWorkflowExecution {
+                            failure: failure.failure,
+                        })],
+                        vec![],
+                        resp_chan,
+                    )
+                    .unwrap_or_else(|e| {
+                        dbg_panic!("Got next page request when auto-failing workflow: {e:?}");
+                        None
+                    });
+            } else {
+                ActivationCompleteOutcome::ReportWFTFail(FailedActivationWFTReport::Report(
+                    tt, cause, failure,
+                ))
+            }
         } else {
             ActivationCompleteOutcome::WFTFailedDontReport
         };
+        self.metrics.wf_task_failed();
         self.reply_to_complete(outcome, resp_chan);
         rur
     }
@@ -1039,6 +1068,25 @@ impl ManagedRun {
                 )
             };
 
+            // Record metrics for any outgoing terminal commands
+            for cmd in commands.iter() {
+                match cmd.attributes.as_ref() {
+                    Some(CmdAttribs::CompleteWorkflowExecutionCommandAttributes(_)) => {
+                        self.metrics.wf_completed();
+                    }
+                    Some(CmdAttribs::FailWorkflowExecutionCommandAttributes(_)) => {
+                        self.metrics.wf_failed();
+                    }
+                    Some(CmdAttribs::ContinueAsNewWorkflowExecutionCommandAttributes(_)) => {
+                        self.metrics.wf_continued_as_new();
+                    }
+                    Some(CmdAttribs::CancelWorkflowExecutionCommandAttributes(_)) => {
+                        self.metrics.wf_canceled();
+                    }
+                    _ => (),
+                }
+            }
+
             ActivationCompleteOutcome::ReportWFTSuccess(ServerCommandsWithWorkflowInfo {
                 task_token: data.task_token,
                 action: ActivationAction::WftComplete {

test-utils/src/lib.rs

@@ -165,7 +165,7 @@ pub struct CoreWfStarter {
     /// Options to use when starting workflow(s)
     pub workflow_options: WorkflowOptions,
     initted_worker: OnceCell<InitializedWorker>,
-    runtime_override: Option<CoreRuntime>,
+    runtime_override: Option<Arc<CoreRuntime>>,
 }
 struct InitializedWorker {
     worker: Arc<dyn CoreWorker>,
@@ -197,7 +197,7 @@ impl CoreWfStarter {
             worker_config,
             initted_worker: OnceCell::new(),
             workflow_options: Default::default(),
-            runtime_override,
+            runtime_override: runtime_override.map(Arc::new),
         }
     }
 
@@ -208,7 +208,7 @@ impl CoreWfStarter {
             task_queue_name: self.task_queue_name.clone(),
             worker_config: self.worker_config.clone(),
             workflow_options: self.workflow_options.clone(),
-            runtime_override: None,
+            runtime_override: self.runtime_override.clone(),
             initted_worker: Default::default(),
         }
     }
@@ -239,6 +239,7 @@ impl CoreWfStarter {
         self.start_wf_with_id(self.task_queue_name.clone()).await
     }
 
+    /// Starts the workflow using the worker, returns run id.
     pub async fn start_with_worker(
         &self,
         wf_name: impl Into<String>,
@@ -496,6 +497,19 @@ impl TestWorker {
         Ok(res)
     }
 
+    pub fn expect_workflow_completion(&self, wf_id: impl Into<String>, run_id: Option<String>) {
+        self.started_workflows.lock().push(WorkflowExecutionInfo {
+            namespace: self
+                .client
+                .as_ref()
+                .map(|c| c.namespace())
+                .unwrap_or(NAMESPACE)
+                .to_owned(),
+            workflow_id: wf_id.into(),
+            run_id,
+        });
+    }
+
     /// Runs until all expected workflows have completed
     pub async fn run_until_done(&mut self) -> Result<(), anyhow::Error> {
         self.run_until_done_intercepted(Option::<TestWorkerCompletionIceptor>::None)

tests/integ_tests/metrics_tests.rs

@@ -34,11 +34,11 @@ use tokio::{join, sync::Barrier, task::AbortHandle};
 
 static ANY_PORT: &str = "127.0.0.1:0";
 
-async fn get_text(endpoint: String) -> String {
+pub async fn get_text(endpoint: String) -> String {
     reqwest::get(endpoint).await.unwrap().text().await.unwrap()
 }
 
-struct AbortOnDrop {
+pub struct AbortOnDrop {
     ah: AbortHandle,
 }
 impl Drop for AbortOnDrop {
@@ -47,7 +47,7 @@ impl Drop for AbortOnDrop {
     }
 }
 
-fn prom_metrics() -> (TelemetryOptions, SocketAddr, AbortOnDrop) {
+pub fn prom_metrics() -> (TelemetryOptions, SocketAddr, AbortOnDrop) {
     let mut telemopts = get_integ_telem_options();
     let prom_info = start_prometheus_metric_exporter(
         PrometheusExporterOptionsBuilder::default()