Support for user metadata in local activity (#974)

Author: maciejdudko

core/src/protosext/mod.rs

@@ -37,6 +37,7 @@ use temporal_sdk_core_protos::{
         failure::v1::Failure,
         history::v1::{History, HistoryEvent, MarkerRecordedEventAttributes, history_event},
         query::v1::WorkflowQuery,
+        sdk::v1::UserMetadata,
         workflowservice::v1::PollWorkflowTaskQueueResponse,
     },
     utilities::TryIntoOrNone,
@@ -320,6 +321,7 @@ pub(crate) struct ValidScheduleLA {
     pub(crate) retry_policy: RetryPolicy,
     pub(crate) local_retry_threshold: Duration,
     pub(crate) cancellation_type: ActivityCancellationType,
+    pub(crate) user_metadata: Option<UserMetadata>,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -349,7 +351,10 @@ impl Default for LACloseTimeouts {
 }
 
 impl ValidScheduleLA {
-    pub(crate) fn from_schedule_la(v: ScheduleLocalActivity) -> Result<Self, anyhow::Error> {
+    pub(crate) fn from_schedule_la(
+        v: ScheduleLocalActivity,
+        user_metadata: Option<UserMetadata>,
+    ) -> Result<Self, anyhow::Error> {
         let original_schedule_time = v
             .original_schedule_time
             .map(|x| {
@@ -423,6 +428,7 @@ impl ValidScheduleLA {
             retry_policy,
             local_retry_threshold,
             cancellation_type,
+            user_metadata,
         })
     }
 }

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

@@ -31,7 +31,7 @@ use temporal_sdk_core_protos::{
         workflow_commands::ActivityCancellationType,
     },
     temporal::api::{
-        command::v1::{RecordMarkerCommandAttributes, command},
+        command::v1::{Command as ProtoCommand, RecordMarkerCommandAttributes, command},
         enums::v1::{CommandType, EventType, RetryState},
         failure::v1::{Failure, failure::FailureInfo},
     },
@@ -772,9 +772,11 @@ impl WFMachinesAdapter for LocalActivityMachine {
                         header: None,
                         failure: maybe_failure,
                     };
-                    responses.push(MachineResponse::IssueNewCommand(
-                        command::Attributes::RecordMarkerCommandAttributes(marker_data).into(),
-                    ));
+                    let command = ProtoCommand {
+                        user_metadata: self.shared_state.attrs.user_metadata.clone(),
+                        ..command::Attributes::RecordMarkerCommandAttributes(marker_data).into()
+                    };
+                    responses.push(MachineResponse::IssueNewCommand(command));
                 }
                 Ok(responses)
             }

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

@@ -1352,7 +1352,7 @@ impl WorkflowMachines {
                 WFCommandVariant::AddLocalActivity(attrs) => {
                     let seq = attrs.seq;
                     let attrs: ValidScheduleLA =
-                        ValidScheduleLA::from_schedule_la(attrs).map_err(|e| {
+                        ValidScheduleLA::from_schedule_la(attrs, cmd.metadata).map_err(|e| {
                             WFMachinesError::Fatal(format!(
                                 "Invalid schedule local activity request (seq {seq}): {e}"
                             ))

sdk/src/workflow_context/options.rs

@@ -3,6 +3,7 @@ use std::{collections::HashMap, time::Duration};
 use temporal_client::{Priority, WorkflowOptions};
 use temporal_sdk_core_protos::{
     coresdk::{
+        AsJsonPayloadExt,
         child_workflow::ChildWorkflowCancellationType,
         nexus::NexusOperationCancellationType,
         workflow_commands::{
@@ -155,6 +156,8 @@ pub struct LocalActivityOptions {
     /// specified. If set, this must be <= `schedule_to_close_timeout`, if not, it will be clamped
     /// down.
     pub start_to_close_timeout: Option<Duration>,
+    /// Single-line summary for this activity that will appear in UI/CLI.
+    pub summary: Option<String>,
 }
 
 impl IntoWorkflowCommand for LocalActivityOptions {
@@ -194,7 +197,13 @@ impl IntoWorkflowCommand for LocalActivityOptions {
                 }
                 .into(),
             ),
-            user_metadata: None,
+            user_metadata: self
+                .summary
+                .and_then(|summary| summary.as_json_payload().ok())
+                .map(|summary| UserMetadata {
+                    summary: Some(summary),
+                    details: None,
+                }),
         }
     }
 }

tests/integ_tests/workflow_tests/local_activities.rs

@@ -19,14 +19,15 @@ use temporal_sdk_core::replay::HistoryForReplay;
 use temporal_sdk_core_protos::{
     TestHistoryBuilder,
     coresdk::{
-        AsJsonPayloadExt, IntoPayloadsExt,
+        AsJsonPayloadExt, FromJsonPayloadExt, IntoPayloadsExt,
         workflow_commands::{ActivityCancellationType, workflow_command::Variant},
         workflow_completion,
         workflow_completion::{WorkflowActivationCompletion, workflow_activation_completion},
     },
     temporal::api::{
         common::v1::RetryPolicy,
         enums::v1::{TimeoutType, UpdateWorkflowExecutionLifecycleStage},
+        history::v1::history_event::Attributes::MarkerRecordedEventAttributes,
         update::v1::WaitPolicy,
     },
 };
@@ -906,3 +907,53 @@ async fn local_activity_with_heartbeat_only_causes_one_wakeup() {
         .unwrap();
     assert_eq!(res[0], replay_res.unwrap());
 }
+
+pub(crate) async fn local_activity_with_summary_wf(ctx: WfContext) -> WorkflowResult<()> {
+    ctx.local_activity(LocalActivityOptions {
+        activity_type: "echo_activity".to_string(),
+        input: "hi!".as_json_payload().expect("serializes fine"),
+        summary: Some("Echo summary".to_string()),
+        ..Default::default()
+    })
+    .await;
+    Ok(().into())
+}
+
+#[tokio::test]
+async fn local_activity_with_summary() {
+    let wf_name = "local_activity_with_summary";
+    let mut starter = CoreWfStarter::new(wf_name);
+    let mut worker = starter.worker().await;
+    worker.register_wf(wf_name.to_owned(), local_activity_with_summary_wf);
+    worker.register_activity("echo_activity", echo);
+
+    let handle = starter.start_with_worker(wf_name, &mut worker).await;
+    worker.run_until_done().await.unwrap();
+    handle
+        .fetch_history_and_replay(worker.inner_mut())
+        .await
+        .unwrap();
+
+    let la_events = starter
+        .get_history()
+        .await
+        .events
+        .into_iter()
+        .filter(|e| match e.attributes {
+            Some(MarkerRecordedEventAttributes(ref a)) => a.marker_name == "core_local_activity",
+            _ => false,
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(la_events.len(), 1);
+    let summary = la_events[0]
+        .user_metadata
+        .as_ref()
+        .expect("metadata missing from local activity marker")
+        .summary
+        .as_ref()
+        .expect("summary missing from local activity marker");
+    assert_eq!(
+        "Echo summary",
+        String::from_json_payload(summary).expect("failed to parse summary")
+    );
+}