Move user metadata to command from variant (#851)

Author: Sushisource

core/src/abstractions.rs

@@ -189,7 +189,7 @@ struct UseCtx<'a, SK: SlotKind> {
     permit: &'a SlotSupplierPermit,
 }
 
-impl<'a, SK: SlotKind> SlotMarkUsedContext for UseCtx<'a, SK> {
+impl<SK: SlotKind> SlotMarkUsedContext for UseCtx<'_, SK> {
     type SlotKind = SK;
 
     fn permit(&self) -> &SlotSupplierPermit {

core/src/core_tests/updates.rs

@@ -291,7 +291,6 @@ async fn replay_with_signal_and_update_same_task() {
             StartTimer {
                 seq: 1,
                 start_to_fire_timeout: Some(prost_dur!(from_secs(1))),
-                summary: None,
             }
             .into(),
             UpdateResponse {

core/src/core_tests/workers.rs

@@ -45,7 +45,6 @@ async fn after_shutdown_of_worker_get_shutdown_err() {
                 workflow_command::Variant::StartTimer(StartTimer {
                     seq: 1,
                     start_to_fire_timeout: Some(prost_dur!(from_secs(1))),
-                    summary: None,
                 }),
             ))
             .await
@@ -349,7 +348,6 @@ async fn worker_shutdown_api(#[case] use_cache: bool, #[case] api_success: bool)
                 workflow_command::Variant::StartTimer(StartTimer {
                     seq: 1,
                     start_to_fire_timeout: Some(prost_dur!(from_secs(1))),
-                    summary: None,
                 }),
             ))
             .await

core/src/telemetry/log_export.rs

@@ -168,7 +168,7 @@ impl fmt::Debug for CoreLogStreamConsumer {
 
 struct JsonVisitor<'a>(&'a mut HashMap<String, serde_json::Value>);
 
-impl<'a> tracing::field::Visit for JsonVisitor<'a> {
+impl tracing::field::Visit for JsonVisitor<'_> {
     fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
         self.0
             .insert(field.name().to_string(), serde_json::json!(value));

core/src/telemetry/mod.rs

@@ -276,11 +276,13 @@ where
     }
 }
 
+/// Helpers for test initialization
 #[cfg(test)]
 pub mod test_initters {
     use super::*;
     use temporal_sdk_core_api::telemetry::TelemetryOptionsBuilder;
 
+    /// Turn on logging to the console
     #[allow(dead_code)] // Not always used, called to enable for debugging when needed
     pub fn test_telem_console() {
         telemetry_init_global(

core/src/test_help/mod.rs

@@ -1026,6 +1026,7 @@ macro_rules! advance_fut {
     };
 }
 
+/// Helps easily construct prost proto durations from stdlib duration constructors
 #[macro_export]
 macro_rules! prost_dur {
     ($dur_call:ident $args:tt) => {

core/src/worker/workflow/driven_workflow.rs

@@ -1,6 +1,6 @@
 use crate::{
     telemetry::VecDisplayer,
-    worker::workflow::{OutgoingJob, WFCommand, WorkflowStartedInfo},
+    worker::workflow::{OutgoingJob, WFCommand, WFCommandVariant, WorkflowStartedInfo},
 };
 use prost_types::Timestamp;
 use std::{
@@ -86,7 +86,12 @@ impl DrivenWorkflow {
     /// from a buffer that the language side sinks into when it calls [crate::Core::complete_task]
     pub(super) fn fetch_workflow_iteration_output(&mut self) -> Vec<WFCommand> {
         let in_cmds = self.incoming_commands.try_recv();
-        let in_cmds = in_cmds.unwrap_or_else(|_| vec![WFCommand::NoCommandsFromLang]);
+        let in_cmds = in_cmds.unwrap_or_else(|_| {
+            vec![WFCommand {
+                variant: WFCommandVariant::NoCommandsFromLang,
+                metadata: None,
+            }]
+        });
         debug!(in_cmds = %in_cmds.display(), "wf bridge iteration fetch");
         in_cmds
     }

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

@@ -30,7 +30,6 @@ use temporal_sdk_core_protos::{
             ActivityTaskCompletedEventAttributes, ActivityTaskFailedEventAttributes,
             ActivityTaskTimedOutEventAttributes,
         },
-        sdk::v1::UserMetadata,
     },
 };
 
@@ -115,10 +114,6 @@ impl ActivityMachine {
         internal_flags: InternalFlagsRef,
         use_compatible_version: bool,
     ) -> NewMachineWithCommand {
-        let user_metadata = attrs.summary.clone().map(|x| UserMetadata {
-            summary: Some(x),
-            details: None,
-        });
         let mut s = Self::from_parts(
             Created {}.into(),
             SharedState {
@@ -133,16 +128,11 @@ impl ActivityMachine {
         );
         OnEventWrapper::on_event_mut(&mut s, ActivityMachineEvents::Schedule)
             .expect("Scheduling activities doesn't fail");
-        let command = Command {
-            command_type: CommandType::ScheduleActivityTask as i32,
-            attributes: Some(schedule_activity_cmd_to_api(
+        NewMachineWithCommand {
+            command: schedule_activity_cmd_to_api(
                 s.shared_state().attrs.clone(),
                 use_compatible_version,
-            )),
-            user_metadata,
-        };
-        NewMachineWithCommand {
-            command,
+            ),
             machine: s.into(),
         }
     }

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

@@ -11,7 +11,7 @@ use temporal_sdk_core_protos::{
         workflow_activation::ResolveRequestCancelExternalWorkflow,
     },
     temporal::api::{
-        command::v1::{command, Command, RequestCancelExternalWorkflowExecutionCommandAttributes},
+        command::v1::{command, RequestCancelExternalWorkflowExecutionCommandAttributes},
         enums::v1::{CancelExternalWorkflowExecutionFailedCause, CommandType, EventType},
         failure::v1::{failure::FailureInfo, ApplicationFailureInfo, Failure},
         history::v1::history_event,
@@ -75,13 +75,8 @@ pub(super) fn new_external_cancel(
             reason,
         },
     );
-    let cmd = Command {
-        command_type: CommandType::RequestCancelExternalWorkflowExecution as i32,
-        attributes: Some(cmd_attrs),
-        user_metadata: Default::default(),
-    };
     NewMachineWithCommand {
-        command: cmd,
+        command: cmd_attrs,
         machine: s.into(),
     }
 }

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

@@ -7,10 +7,7 @@ use rustfsm::{fsm, StateMachine, TransitionResult};
 use std::convert::TryFrom;
 use temporal_sdk_core_protos::{
     coresdk::workflow_commands::CancelWorkflowExecution,
-    temporal::api::{
-        command::v1::Command,
-        enums::v1::{CommandType, EventType},
-    },
+    temporal::api::enums::v1::{CommandType, EventType},
 };
 
 fsm! {
@@ -34,13 +31,8 @@ pub(super) fn cancel_workflow(attribs: CancelWorkflowExecution) -> NewMachineWit
     let mut machine = CancelWorkflowMachine::from_parts(Created {}.into(), ());
     OnEventWrapper::on_event_mut(&mut machine, CancelWorkflowMachineEvents::Schedule)
         .expect("Scheduling continue as new machine doesn't fail");
-    let command = Command {
-        command_type: CommandType::CancelWorkflowExecution as i32,
-        attributes: Some(attribs.into()),
-        user_metadata: Default::default(),
-    };
     NewMachineWithCommand {
-        command,
+        command: attribs.into(),
         machine: machine.into(),
     }
 }