Add wft fail reason to legacy query responses (#950)

Author: Sushisource

client/src/raw.rs

@@ -1336,6 +1336,24 @@ proxier! {
             r.extensions_mut().insert(labels);
         }
     );
+    (
+        list_workers,
+        ListWorkersRequest,
+        ListWorkersResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        record_worker_heartbeat,
+        RecordWorkerHeartbeatRequest,
+        RecordWorkerHeartbeatResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
 }
 
 proxier! {

core/src/core_tests/queries.rs

@@ -3,7 +3,10 @@ use crate::{
         MockPollCfg, MocksHolder, ResponseType, WorkerExt, build_mock_pollers, canned_histories,
         hist_to_poll_resp, mock_worker, single_hist_mock_sg,
     },
-    worker::{LEGACY_QUERY_ID, client::mocks::mock_workflow_client},
+    worker::{
+        LEGACY_QUERY_ID,
+        client::{LegacyQueryResult, mocks::mock_workflow_client},
+    },
 };
 use futures_util::stream;
 use std::{
@@ -25,7 +28,7 @@ use temporal_sdk_core_protos::{
     },
     temporal::api::{
         common::v1::Payload,
-        enums::v1::{CommandType, EventType},
+        enums::v1::{CommandType, EventType, WorkflowTaskFailedCause},
         failure::v1::Failure,
         history::v1::{ActivityTaskCancelRequestedEventAttributes, History, history_event},
         query::v1::WorkflowQuery,
@@ -300,6 +303,12 @@ async fn query_failure_because_nondeterminism(#[values(true, false)] legacy: boo
     }];
     let mut mock = MockPollCfg::from_resp_batches(wfid, t, tasks, mock_workflow_client());
     if legacy {
+        mock.expect_legacy_query_matcher = Box::new(|_, f| match f {
+            LegacyQueryResult::Failed(f) => {
+                f.force_cause() == WorkflowTaskFailedCause::NonDeterministicError
+            }
+            _ => false,
+        });
         mock.num_expected_legacy_query_resps = 1;
     } else {
         mock.num_expected_fails = 1;

core/src/core_tests/workflow_tasks.rs

@@ -2886,6 +2886,7 @@ async fn use_compatible_version_flag(
         VersioningIntent::Compatible => true,
         VersioningIntent::Default => false,
     };
+    #[allow(deprecated)]
     mock_client
         .expect_complete_workflow_task()
         .returning(move |mut c| {

core/src/test_help/mod.rs

@@ -9,7 +9,8 @@ use crate::{
     worker::{
         TaskPollers,
         client::{
-            MockWorkerClient, WorkerClient, WorkflowTaskCompletion, mocks::mock_workflow_client,
+            LegacyQueryResult, MockWorkerClient, WorkerClient, WorkflowTaskCompletion,
+            mocks::mock_workflow_client,
         },
     },
 };
@@ -406,6 +407,9 @@ pub(crate) struct MockPollCfg {
     /// All calls to fail WFTs must match this predicate
     pub(crate) expect_fail_wft_matcher:
         Box<dyn Fn(&TaskToken, &WorkflowTaskFailedCause, &Option<Failure>) -> bool + Send>,
+    /// All calls to legacy query responses must match this predicate
+    pub(crate) expect_legacy_query_matcher:
+        Box<dyn Fn(&TaskToken, &LegacyQueryResult) -> bool + Send>,
     pub(crate) completion_mock_fn: Option<Box<WFTCompletionMockFn>>,
     pub(crate) num_expected_completions: Option<TimesRange>,
     /// If being used with the Rust SDK, this is set true. It ensures pollers will not error out
@@ -428,6 +432,7 @@ impl MockPollCfg {
             num_expected_legacy_query_resps: 0,
             mock_client: mock_workflow_client(),
             expect_fail_wft_matcher: Box::new(|_, _, _| true),
+            expect_legacy_query_matcher: Box::new(|_, _| true),
             completion_mock_fn: None,
             num_expected_completions: None,
             using_rust_sdk: false,
@@ -466,6 +471,7 @@ impl MockPollCfg {
             num_expected_legacy_query_resps: 0,
             mock_client,
             expect_fail_wft_matcher: Box::new(|_, _, _| true),
+            expect_legacy_query_matcher: Box::new(|_, _| true),
             completion_mock_fn: None,
             num_expected_completions: None,
             using_rust_sdk: false,
@@ -709,6 +715,7 @@ pub(crate) fn build_mock_pollers(mut cfg: MockPollCfg) -> MocksHolder {
     let outstanding = outstanding_wf_task_tokens.clone();
     cfg.mock_client
         .expect_respond_legacy_query()
+        .withf(cfg.expect_legacy_query_matcher)
         .times::<TimesRange>(cfg.num_expected_legacy_query_resps.into())
         .returning(move |tt, _| {
             outstanding.release_token(&tt);

core/src/worker/client.rs

@@ -1,6 +1,7 @@
 //! Worker-specific client needs
 
 pub(crate) mod mocks;
+use crate::protosext::legacy_query_failure;
 use parking_lot::RwLock;
 use std::{sync::Arc, time::Duration};
 use temporal_client::{
@@ -10,7 +11,7 @@ use temporal_client::{
 use temporal_sdk_core_api::worker::WorkerVersioningStrategy;
 use temporal_sdk_core_protos::{
     TaskToken,
-    coresdk::workflow_commands::QueryResult,
+    coresdk::{workflow_commands::QueryResult, workflow_completion},
     temporal::api::{
         command::v1::Command,
         common::v1::{
@@ -34,6 +35,11 @@ use tonic::IntoRequest;
 
 type Result<T, E = tonic::Status> = std::result::Result<T, E>;
 
+pub enum LegacyQueryResult {
+    Succeeded(QueryResult),
+    Failed(workflow_completion::Failure),
+}
+
 /// Contains everything a worker needs to interact with the server
 pub(crate) struct WorkerClientBag {
     replaceable_client: RwLock<RetryClient<Client>>,
@@ -197,7 +203,7 @@ pub trait WorkerClient: Sync + Send {
     async fn respond_legacy_query(
         &self,
         task_token: TaskToken,
-        query_result: QueryResult,
+        query_result: LegacyQueryResult,
     ) -> Result<RespondQueryTaskCompletedResponse>;
     /// Describe the namespace
     async fn describe_namespace(&self) -> Result<DescribeNamespaceResponse>;
@@ -267,6 +273,7 @@ impl WorkerClient for WorkerClientBag {
             binary_checksum: self.binary_checksum(),
             worker_version_capabilities: self.worker_version_capabilities(),
             deployment_options: self.deployment_options(),
+            worker_heartbeat: None,
         }
         .into_request();
         request.extensions_mut().insert(IsWorkerTaskLongPoll);
@@ -303,6 +310,7 @@ impl WorkerClient for WorkerClientBag {
             }),
             worker_version_capabilities: self.worker_version_capabilities(),
             deployment_options: self.deployment_options(),
+            worker_heartbeat: None,
         }
         .into_request();
         request.extensions_mut().insert(IsWorkerTaskLongPoll);
@@ -335,6 +343,7 @@ impl WorkerClient for WorkerClientBag {
             identity: self.identity.clone(),
             worker_version_capabilities: self.worker_version_capabilities(),
             deployment_options: self.deployment_options(),
+            worker_heartbeat: None,
         }
         .into_request();
         request.extensions_mut().insert(IsWorkerTaskLongPoll);
@@ -578,9 +587,20 @@ impl WorkerClient for WorkerClientBag {
     async fn respond_legacy_query(
         &self,
         task_token: TaskToken,
-        query_result: QueryResult,
+        query_result: LegacyQueryResult,
     ) -> Result<RespondQueryTaskCompletedResponse> {
+        let mut failure = None;
+        let (query_result, cause) = match query_result {
+            LegacyQueryResult::Succeeded(s) => (s, WorkflowTaskFailedCause::Unspecified),
+            #[allow(deprecated)]
+            LegacyQueryResult::Failed(f) => {
+                let cause = f.force_cause();
+                failure = f.failure.clone();
+                (legacy_query_failure(f), cause)
+            }
+        };
         let (_, completed_type, query_result, error_message) = query_result.into_components();
+
         Ok(self
             .cloned_client()
             .respond_query_task_completed(RespondQueryTaskCompletedRequest {
@@ -589,8 +609,8 @@ impl WorkerClient for WorkerClientBag {
                 query_result,
                 error_message,
                 namespace: self.namespace.clone(),
-                // TODO: https://github.com/temporalio/sdk-core/issues/867
-                failure: None,
+                failure,
+                cause: cause.into(),
             })
             .await?
             .into_inner())
@@ -612,6 +632,7 @@ impl WorkerClient for WorkerClientBag {
             identity: self.identity.clone(),
             sticky_task_queue,
             reason: "graceful shutdown".to_string(),
+            worker_heartbeat: None,
         };
 
         Ok(

core/src/worker/client/mocks.rs

@@ -135,7 +135,7 @@ mockall::mock! {
         fn respond_legacy_query<'a, 'b>(
             &self,
             task_token: TaskToken,
-            query_result: QueryResult,
+        query_result: LegacyQueryResult,
         ) -> impl Future<Output = Result<RespondQueryTaskCompletedResponse>> + Send + 'b
             where 'a: 'b, Self: 'b;
 

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

@@ -69,10 +69,9 @@ pub(super) fn new_external_cancel(
             namespace: workflow_execution.namespace,
             workflow_id: workflow_execution.workflow_id,
             run_id: workflow_execution.run_id,
-            // Apparently this is effectively deprecated at this point
-            control: "".to_string(),
             child_workflow_only: only_child,
             reason,
+            ..Default::default()
         },
     );
     NewMachineWithCommand {

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

@@ -714,7 +714,7 @@ impl WFMachinesAdapter for ChildWorkflowMachine {
                             run_id: self.shared_state.run_id.clone(),
                             child_workflow_only: true,
                             reason,
-                            control: "".to_string(),
+                            ..Default::default()
                         }
                         .into(),
                     ))

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

@@ -628,8 +628,8 @@ impl NexusOperationMachine {
                     endpoint: self.shared_state.endpoint.clone(),
                     service: self.shared_state.service.clone(),
                     operation: self.shared_state.operation.clone(),
-                    operation_id: "".to_string(),
                     operation_token: operation_token.clone().unwrap_or_default(),
+                    ..Default::default()
                 },
             )),
             ..Default::default()

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

@@ -104,9 +104,8 @@ pub(super) fn new_external_signal(
             },
             signal_name: attrs.signal_name,
             input: attrs.args.into_payloads(),
-            // Is deprecated
-            control: "".to_string(),
             child_workflow_only: only_child,
+            ..Default::default()
         },
     );
     Ok(NewMachineWithCommand {