Support user metadata in child workflow, timer, and activity

core/src/core_tests/activity_tasks.rs

@@ -2,9 +2,9 @@ use crate::{
     advance_fut, job_assert, prost_dur,
     test_help::{
         build_fake_worker, build_mock_pollers, canned_histories, gen_assert_and_reply,
-        mock_manual_poller, mock_poller, mock_poller_from_resps, mock_worker, poll_and_reply,
-        single_hist_mock_sg, test_worker_cfg, MockPollCfg, MockWorkerInputs, MocksHolder,
-        QueueResponse, ResponseType, WorkerExt, WorkflowCachingPolicy, TEST_Q,
+        mock_manual_poller, mock_poller, mock_poller_from_resps, mock_sdk_cfg, mock_worker,
+        poll_and_reply, single_hist_mock_sg, test_worker_cfg, MockPollCfg, MockWorkerInputs,
+        MocksHolder, QueueResponse, ResponseType, WorkerExt, WorkflowCachingPolicy, TEST_Q,
     },
     worker::client::mocks::{mock_manual_workflow_client, mock_workflow_client},
     ActivityHeartbeat, Worker,
@@ -45,17 +45,18 @@ use temporal_sdk_core_protos::{
     },
     temporal::api::{
         command::v1::{command::Attributes, ScheduleActivityTaskCommandAttributes},
-        enums::v1::EventType,
+        enums::v1::{CommandType, EventType},
         history::v1::{
             history_event::Attributes as EventAttributes, ActivityTaskScheduledEventAttributes,
         },
+        sdk::v1::UserMetadata,
         workflowservice::v1::{
             PollActivityTaskQueueResponse, RecordActivityTaskHeartbeatResponse,
             RespondActivityTaskCanceledResponse, RespondActivityTaskCompletedResponse,
             RespondActivityTaskFailedResponse, RespondWorkflowTaskCompletedResponse,
         },
     },
-    TestHistoryBuilder, DEFAULT_WORKFLOW_TYPE,
+    TestHistoryBuilder, DEFAULT_ACTIVITY_TYPE, DEFAULT_WORKFLOW_TYPE,
 };
 use temporal_sdk_core_test_utils::{fanout_tasks, start_timer_cmd, TestWorker};
 use tokio::{join, sync::Barrier, time::sleep};
@@ -1187,3 +1188,54 @@ async fn activities_must_be_flushed_to_server_on_shutdown(#[values(true, false)]
     };
     join!(shutdown_task, complete_task);
 }
+
+#[tokio::test]
+async fn pass_activity_summary_to_metadata() {
+    let t = canned_histories::single_activity("1");
+    let mut mock_cfg = MockPollCfg::from_hist_builder(t);
+    let wf_id = mock_cfg.hists[0].wf_id.clone();
+    let wf_type = DEFAULT_WORKFLOW_TYPE;
+    let expected_user_metadata = Some(UserMetadata {
+        summary: Some(b"activity summary".into()),
+        details: None,
+    });
+    mock_cfg.completion_asserts_from_expectations(|mut asserts| {
+        asserts
+            .then(move |wft| {
+                assert_eq!(wft.commands.len(), 1);
+                assert_eq!(
+                    wft.commands[0].command_type(),
+                    CommandType::ScheduleActivityTask
+                );
+                assert_eq!(wft.commands[0].user_metadata, expected_user_metadata)
+            })
+            .then(move |wft| {
+                assert_eq!(wft.commands.len(), 1);
+                assert_eq!(
+                    wft.commands[0].command_type(),
+                    CommandType::CompleteWorkflowExecution
+                );
+            });
+    });
+
+    let mut worker = mock_sdk_cfg(mock_cfg, |_| {});
+    worker.register_wf(wf_type, |ctx: WfContext| async move {
+        ctx.activity(ActivityOptions {
+            activity_type: DEFAULT_ACTIVITY_TYPE.to_string(),
+            summary: Some("activity summary".to_string()),
+            ..Default::default()
+        })
+        .await;
+        Ok(().into())
+    });
+    worker
+        .submit_wf(
+            wf_id.to_owned(),
+            wf_type.to_owned(),
+            vec![],
+            WorkflowOptions::default(),
+        )
+        .await
+        .unwrap();
+    worker.run_until_done().await.unwrap();
+}

core/src/core_tests/child_workflows.rs

@@ -1,21 +1,24 @@
 use crate::{
     replay::DEFAULT_WORKFLOW_TYPE,
     test_help::{
-        build_fake_sdk, canned_histories, mock_sdk, mock_worker, single_hist_mock_sg, MockPollCfg,
-        ResponseType,
+        build_fake_sdk, canned_histories, mock_sdk, mock_sdk_cfg, mock_worker, single_hist_mock_sg,
+        MockPollCfg, ResponseType,
     },
     worker::client::mocks::mock_workflow_client,
 };
 use temporal_client::WorkflowOptions;
 use temporal_sdk::{ChildWorkflowOptions, Signal, WfContext, WorkflowResult};
 use temporal_sdk_core_api::Worker;
-use temporal_sdk_core_protos::coresdk::{
-    child_workflow::{child_workflow_result, ChildWorkflowCancellationType},
-    workflow_activation::{workflow_activation_job, WorkflowActivationJob},
-    workflow_commands::{
-        CancelChildWorkflowExecution, CompleteWorkflowExecution, StartChildWorkflowExecution,
+use temporal_sdk_core_protos::{
+    coresdk::{
+        child_workflow::{child_workflow_result, ChildWorkflowCancellationType},
+        workflow_activation::{workflow_activation_job, WorkflowActivationJob},
+        workflow_commands::{
+            CancelChildWorkflowExecution, CompleteWorkflowExecution, StartChildWorkflowExecution,
+        },
+        workflow_completion::WorkflowActivationCompletion,
     },
-    workflow_completion::WorkflowActivationCompletion,
+    temporal::api::{enums::v1::CommandType, sdk::v1::UserMetadata},
 };
 use tokio::join;
 
@@ -220,3 +223,57 @@ async fn cancel_already_complete_child_ignored() {
     .await
     .unwrap();
 }
+
+#[tokio::test]
+async fn pass_child_workflow_summary_to_metadata() {
+    let wf_id = "1";
+    let wf_type = DEFAULT_WORKFLOW_TYPE;
+    let t = canned_histories::single_child_workflow(wf_id);
+    let mut mock_cfg = MockPollCfg::from_hist_builder(t);
+    let expected_user_metadata = Some(UserMetadata {
+        summary: Some(b"child summary".into()),
+        details: Some(b"child details".into()),
+    });
+    mock_cfg.completion_asserts_from_expectations(|mut asserts| {
+        asserts
+            .then(move |wft| {
+                assert_eq!(wft.commands.len(), 1);
+                assert_eq!(
+                    wft.commands[0].command_type(),
+                    CommandType::StartChildWorkflowExecution
+                );
+                assert_eq!(wft.commands[0].user_metadata, expected_user_metadata)
+            })
+            .then(move |wft| {
+                assert_eq!(wft.commands.len(), 1);
+                assert_eq!(
+                    wft.commands[0].command_type(),
+                    CommandType::CompleteWorkflowExecution
+                );
+            });
+    });
+
+    let mut worker = mock_sdk_cfg(mock_cfg, |_| {});
+    worker.register_wf(wf_type, move |ctx: WfContext| async move {
+        ctx.child_workflow(ChildWorkflowOptions {
+            workflow_id: wf_id.to_string(),
+            workflow_type: "child".to_string(),
+            static_summary: Some("child summary".to_string()),
+            static_details: Some("child details".to_string()),
+            ..Default::default()
+        })
+        .start(&ctx)
+        .await;
+        Ok(().into())
+    });
+    worker
+        .submit_wf(
+            wf_id.to_owned(),
+            wf_type.to_owned(),
+            vec![],
+            WorkflowOptions::default(),
+        )
+        .await
+        .unwrap();
+    worker.run_until_done().await.unwrap();
+}

core/src/core_tests/updates.rs

@@ -291,6 +291,7 @@ 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,6 +45,7 @@ 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
@@ -352,6 +353,7 @@ 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/core_tests/workflow_tasks.rs

@@ -29,7 +29,7 @@ use std::{
     time::Duration,
 };
 use temporal_client::WorkflowOptions;
-use temporal_sdk::{ActivityOptions, CancellableFuture, WfContext};
+use temporal_sdk::{ActivityOptions, CancellableFuture, TimerOptions, WfContext};
 use temporal_sdk_core_api::{
     errors::PollWfError,
     worker::{
@@ -64,6 +64,7 @@ use temporal_sdk_core_protos::{
             history_event, TimerFiredEventAttributes,
             WorkflowPropertiesModifiedExternallyEventAttributes,
         },
+        sdk::v1::UserMetadata,
         workflowservice::v1::{
             GetWorkflowExecutionHistoryResponse, RespondWorkflowTaskCompletedResponse,
         },
@@ -3086,3 +3087,50 @@ async fn slot_provider_cant_hand_out_more_permits_than_cache_size() {
     // polling is not exceeding the task limit
     assert_eq!(popped_tasks.load(Ordering::Relaxed), 10);
 }
+
+#[tokio::test]
+async fn pass_timer_summary_to_metadata() {
+    let t = canned_histories::single_timer("1");
+    let mut mock_cfg = MockPollCfg::from_hist_builder(t);
+    let wf_id = mock_cfg.hists[0].wf_id.clone();
+    let wf_type = DEFAULT_WORKFLOW_TYPE;
+    let expected_user_metadata = Some(UserMetadata {
+        summary: Some(b"timer summary".into()),
+        details: None,
+    });
+    mock_cfg.completion_asserts_from_expectations(|mut asserts| {
+        asserts
+            .then(move |wft| {
+                assert_eq!(wft.commands.len(), 1);
+                assert_eq!(wft.commands[0].command_type(), CommandType::StartTimer);
+                assert_eq!(wft.commands[0].user_metadata, expected_user_metadata)
+            })
+            .then(move |wft| {
+                assert_eq!(wft.commands.len(), 1);
+                assert_eq!(
+                    wft.commands[0].command_type(),
+                    CommandType::CompleteWorkflowExecution
+                );
+            });
+    });
+
+    let mut worker = mock_sdk_cfg(mock_cfg, |_| {});
+    worker.register_wf(wf_type, |ctx: WfContext| async move {
+        ctx.timer(TimerOptions {
+            duration: Duration::from_secs(1),
+            summary: Some("timer summary".to_string()),
+        })
+        .await;
+        Ok(().into())
+    });
+    worker
+        .submit_wf(
+            wf_id.to_owned(),
+            wf_type.to_owned(),
+            vec![],
+            WorkflowOptions::default(),
+        )
+        .await
+        .unwrap();
+    worker.run_until_done().await.unwrap();
+}

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

@@ -30,6 +30,7 @@ use temporal_sdk_core_protos::{
             ActivityTaskCompletedEventAttributes, ActivityTaskFailedEventAttributes,
             ActivityTaskTimedOutEventAttributes,
         },
+        sdk::v1::UserMetadata,
     },
 };
 
@@ -114,6 +115,10 @@ 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 {
@@ -134,7 +139,7 @@ impl ActivityMachine {
                 s.shared_state().attrs.clone(),
                 use_compatible_version,
             )),
-            user_metadata: Default::default(),
+            user_metadata,
         };
         NewMachineWithCommand {
             command,

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

@@ -41,6 +41,7 @@ use temporal_sdk_core_protos::{
             ChildWorkflowExecutionStartedEventAttributes,
             StartChildWorkflowExecutionFailedEventAttributes,
         },
+        sdk::v1::UserMetadata,
     },
 };
 
@@ -449,6 +450,15 @@ impl ChildWorkflowMachine {
                 cancelled_before_sent: false,
             },
         );
+        let user_metadata = if attribs.static_summary.is_some() || attribs.static_details.is_some()
+        {
+            Some(UserMetadata {
+                summary: attribs.static_summary.clone(),
+                details: attribs.static_details.clone(),
+            })
+        } else {
+            None
+        };
         OnEventWrapper::on_event_mut(&mut s, ChildWorkflowMachineEvents::Schedule)
             .expect("Scheduling child workflows doesn't fail");
         let cmd = Command {
@@ -457,7 +467,7 @@ impl ChildWorkflowMachine {
                 attribs,
                 use_compatible_version,
             )),
-            user_metadata: Default::default(),
+            user_metadata,
         };
         NewMachineWithCommand {
             command: cmd,

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

@@ -17,6 +17,7 @@ use temporal_sdk_core_protos::{
         command::v1::Command,
         enums::v1::{CommandType, EventType},
         history::v1::{history_event, TimerFiredEventAttributes},
+        sdk::v1::UserMetadata,
     },
 };
 
@@ -73,13 +74,17 @@ pub(super) fn new_timer(attribs: StartTimer) -> NewMachineWithCommand {
 impl TimerMachine {
     /// Create a new timer and immediately schedule it
     fn new_scheduled(attribs: StartTimer) -> (Self, Command) {
+        let user_metadata = attribs.summary.clone().map(|x| UserMetadata {
+            summary: Some(x),
+            details: None,
+        });
         let mut s = Self::new(attribs);
         OnEventWrapper::on_event_mut(&mut s, TimerMachineEvents::Schedule)
             .expect("Scheduling timers doesn't fail");
         let cmd = Command {
             command_type: CommandType::StartTimer as i32,
-            attributes: Some(s.shared_state().attrs.into()),
-            user_metadata: Default::default(),
+            attributes: Some(s.shared_state().attrs.clone().into()),
+            user_metadata,
         };
         (s, cmd)
     }

sdk-core-protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto

@@ -46,6 +46,8 @@ message StartTimer {
     // Lang's incremental sequence number, used as the operation identifier
     uint32 seq = 1;
     google.protobuf.Duration start_to_fire_timeout = 2;
+    // Summary that is stored as user_metadata
+    temporal.api.common.v1.Payload summary = 3;
 }
 
 message CancelTimer {
@@ -88,6 +90,8 @@ message ScheduleActivity {
     bool do_not_eagerly_execute = 14;
     // Whether this activity should run on a worker with a compatible build id or not.
     coresdk.common.VersioningIntent versioning_intent = 15;
+    // Summary that is stored as user_metadata
+    temporal.api.common.v1.Payload summary = 16;
 }
 
 message ScheduleLocalActivity {
@@ -253,6 +257,10 @@ message StartChildWorkflowExecution {
     child_workflow.ChildWorkflowCancellationType cancellation_type = 18;
     // Whether this child should run on a worker with a compatible build id or not.
     coresdk.common.VersioningIntent versioning_intent = 19;
+    // Static summary of the child workflow
+    temporal.api.common.v1.Payload static_summary = 20;
+    // Static details of the child workflow
+    temporal.api.common.v1.Payload static_details = 21;
 }
 
 // Cancel a child workflow

sdk/src/lib.rs

@@ -57,7 +57,7 @@ use tracing::{field, Instrument, Span};
 pub use workflow_context::{
     ActivityOptions, CancellableFuture, ChildWorkflow, ChildWorkflowOptions, LocalActivityOptions,
     PendingChildWorkflow, Signal, SignalData, SignalWorkflowOptions, StartedChildWorkflow,
-    WfContext,
+    TimerOptions, WfContext,
 };
 
 use crate::{interceptors::WorkerInterceptor, workflow_context::ChildWfCommon};