Merge branch 'master' into optimize-deps

Sushisource · web-flow · commit cda1fdb381d1 · 2026-04-16T10:12:40.000-07:00
diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs
@@ -71,9 +71,12 @@ use std::{
 };
 use temporalio_common::{
     HasWorkflowDefinition,
-    data_converters::{DataConverter, SerializationContextData},
+    data_converters::{
+        DataConverter, GenericPayloadConverter, PayloadConverter, SerializationContext,
+        SerializationContextData,
+    },
     protos::{
-        coresdk::{AsJsonPayloadExt, IntoPayloadsExt},
+        coresdk::IntoPayloadsExt,
         grpc::health::v1::health_client::HealthClient,
         proto_ts_to_system_time,
         temporal::api::{
@@ -1063,13 +1066,20 @@ where
 
         let user_metadata = if options.static_summary.is_some() || options.static_details.is_some()
         {
+            let payload_converter = PayloadConverter::default();
+            let context = SerializationContext {
+                data: &SerializationContextData::Workflow,
+                converter: &payload_converter,
+            };
             Some(UserMetadata {
                 summary: options.static_summary.map(|s| {
-                    s.as_json_payload()
+                    payload_converter
+                        .to_payload(&context, &s)
                         .expect("String-to-JSON payload serialization is infallible")
                 }),
                 details: options.static_details.map(|s| {
-                    s.as_json_payload()
+                    payload_converter
+                        .to_payload(&context, &s)
                         .expect("String-to-JSON payload serialization is infallible")
                 }),
             })
diff --git a/crates/client/src/workflow_handle.rs b/crates/client/src/workflow_handle.rs
@@ -12,7 +12,10 @@ use std::{fmt::Debug, marker::PhantomData};
 pub use temporalio_common::UntypedWorkflow;
 use temporalio_common::{
     HasWorkflowDefinition, QueryDefinition, SignalDefinition, UpdateDefinition, WorkflowDefinition,
-    data_converters::{DataConverter, PayloadConversionError, RawValue, SerializationContextData},
+    data_converters::{
+        DataConverter, GenericPayloadConverter, PayloadConversionError, PayloadConverter, RawValue,
+        SerializationContext, SerializationContextData,
+    },
     payload_visitor::decode_payloads,
     protos::{
         coresdk::FromPayloadsExt,
@@ -48,21 +51,25 @@ struct DecodedUserMetadata {
     details: Option<String>,
 }
 
-async fn decode_user_metadata(
-    data_converter: &DataConverter,
+fn decode_user_metadata(
     context: &SerializationContextData,
     user_metadata: Option<UserMetadata>,
 ) -> Result<DecodedUserMetadata, PayloadConversionError> {
+    let payload_converter = PayloadConverter::default();
+    let context = SerializationContext {
+        data: context,
+        converter: &payload_converter,
+    };
     let (summary, details) = user_metadata
         .map(|metadata| (metadata.summary, metadata.details))
         .unwrap_or_default();
     Ok(DecodedUserMetadata {
         summary: match summary {
-            Some(payload) => Some(data_converter.from_payload(context, payload).await?),
+            Some(payload) => Some(payload_converter.from_payload(&context, payload)?),
             None => None,
         },
         details: match details {
-            Some(payload) => Some(data_converter.from_payload(context, payload).await?),
+            Some(payload) => Some(payload_converter.from_payload(&context, payload)?),
             None => None,
         },
     })
@@ -119,12 +126,8 @@ impl WorkflowExecutionDescription {
             &SerializationContextData::Workflow,
         )
         .await;
-        let decoded_metadata = decode_user_metadata(
-            data_converter,
-            &SerializationContextData::Workflow,
-            raw_user_metadata,
-        )
-        .await?;
+        let decoded_metadata =
+            decode_user_metadata(&SerializationContextData::Workflow, raw_user_metadata)?;
         let history_length_raw = raw_description
             .workflow_execution_info
             .as_ref()
diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml
@@ -68,7 +68,7 @@ opentelemetry-otlp = { version = "0.31", default-features = false, features = [
   "grpc-tonic",
   "reqwest-rustls",
 ], optional = true }
-parking_lot = { version = "0.12", features = ["send_guard"] }
+parking_lot = { version = "0.12" }
 prometheus = { version = "0.14", optional = true, default-features = false }
 prost = { workspace = true }
 prost-wkt = "0.7"
diff --git a/crates/sdk-core/Cargo.toml b/crates/sdk-core/Cargo.toml
@@ -64,7 +64,7 @@ opentelemetry-otlp = { version = "0.31", default-features = false, features = [
   "grpc-tonic",
   "reqwest-rustls",
 ], optional = true }
-parking_lot = { version = "0.12", features = ["send_guard"] }
+parking_lot = { version = "0.12" }
 pid = "4.0"
 pin-project = "1.1"
 prost = { workspace = true }
diff --git a/crates/sdk-core/tests/integ_tests/client_tests.rs b/crates/sdk-core/tests/integ_tests/client_tests.rs
@@ -375,7 +375,7 @@ async fn http_proxy() {
     opts.set_skip_get_system_info(true);
 
     // Connect client with no proxy and make call and confirm reached
-    opts.target = format!("http://127.0.0.1:{}", server.addr.port())
+    opts.target = format!("http://[::1]:{}", server.addr.port())
         .parse()
         .unwrap();
     let connection = Connection::connect(opts.clone()).await.unwrap();
diff --git a/crates/sdk-core/tests/integ_tests/data_converter_tests.rs b/crates/sdk-core/tests/integ_tests/data_converter_tests.rs
@@ -284,19 +284,30 @@ async fn multi_args_serializes_as_multiple_payloads() {
 /// A codec that XORs payload data with a key and tracks encode/decode operations.
 struct XorCodec {
     key: u8,
+    gate_on_metadata: bool,
     encode_count: AtomicUsize,
     decode_count: AtomicUsize,
 }
 
 impl XorCodec {
     fn new(key: u8) -> Self {
         Self {
+            gate_on_metadata: true,
             key,
             encode_count: AtomicUsize::new(0),
             decode_count: AtomicUsize::new(0),
         }
     }
 
+    fn new_with_metadata_gate(key: u8, gate_on_metadata: bool) -> Self {
+        Self {
+            key,
+            gate_on_metadata,
+            encode_count: AtomicUsize::new(0),
+            decode_count: AtomicUsize::new(0),
+        }
+    }
+
     fn encode_count(&self) -> usize {
         self.encode_count.load(Ordering::SeqCst)
     }
@@ -316,12 +327,15 @@ impl PayloadCodec for XorCodec {
         eprintln!("XorCodec::encode called with {} payloads", count);
         self.encode_count.fetch_add(count, Ordering::SeqCst);
         let key = self.key;
+        let gate_on_metadata = self.gate_on_metadata;
         async move {
             payloads
                 .into_iter()
                 .map(|mut p| {
                     p.data = p.data.iter().map(|b| b ^ key).collect();
-                    p.metadata.insert("xor_encoded".to_string(), vec![key]);
+                    if gate_on_metadata {
+                        p.metadata.insert("xor_encoded".to_string(), vec![key]);
+                    }
                     p
                 })
                 .collect()
@@ -338,11 +352,12 @@ impl PayloadCodec for XorCodec {
         eprintln!("XorCodec::decode called with {} payloads", count);
         self.decode_count.fetch_add(count, Ordering::SeqCst);
         let key = self.key;
+        let gate_on_metadata = self.gate_on_metadata;
         async move {
             payloads
                 .into_iter()
                 .map(|mut p| {
-                    if p.metadata.remove("xor_encoded").is_some() {
+                    if !gate_on_metadata || p.metadata.remove("xor_encoded").is_some() {
                         p.data = p.data.iter().map(|b| b ^ key).collect();
                     }
                     p
@@ -464,6 +479,77 @@ async fn describe_decodes_workflow_payload_fields() {
         desc.memo().unwrap().fields["tracked"],
         "codec-describe".as_json_payload().unwrap()
     );
+    let raw_user_metadata = desc
+        .raw_description
+        .execution_config
+        .as_ref()
+        .and_then(|cfg| cfg.user_metadata.as_ref())
+        .expect("describe response should include user metadata");
+    assert_eq!(
+        raw_user_metadata.summary,
+        Some("codec summary".as_json_payload().unwrap())
+    );
+    assert_eq!(
+        raw_user_metadata.details,
+        Some("codec details".as_json_payload().unwrap())
+    );
+    assert_eq!(desc.static_summary(), Some("codec summary"));
+    assert_eq!(desc.static_details(), Some("codec details"));
+}
+
+#[tokio::test]
+async fn describe_decodes_user_metadata_with_ungated_xor_codec() {
+    let wf_name = DescribeDataConverterWorkflow::name();
+    let codec = Arc::new(XorCodec::new_with_metadata_gate(0x42, false));
+
+    let connection = get_integ_connection(None).await;
+    let data_converter = DataConverter::new(
+        PayloadConverter::default(),
+        DefaultFailureConverter,
+        codec.clone(),
+    );
+    let client_opts = ClientOptions::new(integ_namespace())
+        .data_converter(data_converter)
+        .build();
+    let client = Client::new(connection, client_opts).unwrap();
+
+    let mut starter = CoreWfStarter::new_with_overrides(wf_name, None, Some(client));
+    starter.sdk_config.register_activities(TestActivities);
+    starter.sdk_config.task_types = WorkerTaskTypes::all();
+    starter
+        .sdk_config
+        .register_workflow::<DescribeDataConverterWorkflow>();
+    let wf_id = starter.get_task_queue().to_owned();
+    let mut worker = starter.worker().await;
+
+    let handle = worker
+        .submit_workflow(
+            DescribeDataConverterWorkflow::run,
+            TrackedWrapper(TrackedValue::new("codec-describe".to_string())),
+            WorkflowStartOptions::new(starter.get_task_queue(), wf_id)
+                .static_summary("codec summary")
+                .static_details("codec details")
+                .build(),
+        )
+        .await
+        .unwrap();
+    worker.run_until_done().await.unwrap();
+
+    let decode_count_before = codec.decode_count();
+    let desc = handle
+        .describe(WorkflowDescribeOptions::default())
+        .await
+        .unwrap();
+
+    assert!(
+        codec.decode_count() > decode_count_before,
+        "Describe should have decoded response payloads"
+    );
+    assert_eq!(
+        desc.memo().unwrap().fields["tracked"],
+        "codec-describe".as_json_payload().unwrap()
+    );
+    // Making sure codec isn't used when decoding user metadata
     assert_eq!(desc.static_summary(), Some("codec summary"));
     assert_eq!(desc.static_details(), Some("codec details"));
 }
diff --git a/crates/sdk-core/tests/integ_tests/workflow_tests.rs b/crates/sdk-core/tests/integ_tests/workflow_tests.rs
@@ -44,7 +44,7 @@ use temporalio_common::{
     protos::{
         DEFAULT_WORKFLOW_TYPE, canned_histories,
         coresdk::{
-            ActivityTaskCompletion, IntoCompletion,
+            ActivityTaskCompletion, AsJsonPayloadExt, IntoCompletion,
             activity_result::ActivityExecutionResult,
             workflow_activation::{WorkflowActivationJob, workflow_activation_job},
             workflow_commands::{
@@ -1061,7 +1061,7 @@ async fn pass_timer_summary_to_metadata() {
     let mut mock_cfg = MockPollCfg::from_hist_builder(t);
     let wf_id = mock_cfg.hists[0].wf_id.clone();
     let expected_user_metadata = Some(UserMetadata {
-        summary: Some(b"timer summary".into()),
+        summary: Some("timer summary".as_json_payload().unwrap()),
         details: None,
     });
     mock_cfg.completion_asserts_from_expectations(|mut asserts| {
diff --git a/crates/sdk/Cargo.toml b/crates/sdk/Cargo.toml
@@ -21,7 +21,7 @@ futures-util = { version = "0.3", default-features = false, features = [
   "async-await-macro",
 ] }
 gethostname = "1.0.2"
-parking_lot = { version = "0.12", features = ["send_guard"] }
+parking_lot = { version = "0.12" }
 prost-types = { workspace = true }
 serde = "1.0"
 thiserror = "2"
diff --git a/crates/sdk/src/workflow_context.rs b/crates/sdk/src/workflow_context.rs
@@ -421,6 +421,11 @@ impl BaseWorkflowContext {
         let seq = self.inner.seq_nums.borrow_mut().next_timer_seq();
         let (cmd, unblocker) =
             CancellableWFCommandFut::new(CancellableID::Timer(seq), self.clone());
+        let payload_converter = PayloadConverter::default();
+        let context = SerializationContext {
+            data: &SerializationContextData::Workflow,
+            converter: &payload_converter,
+        };
         self.send(
             CommandCreateRequest {
                 cmd: WorkflowCommand {
@@ -436,7 +441,11 @@ impl BaseWorkflowContext {
                         .into(),
                     ),
                     user_metadata: Some(UserMetadata {
-                        summary: opts.summary.map(|x| x.as_bytes().into()),
+                        summary: opts.summary.map(|summary| {
+                            payload_converter
+                                .to_payload(&context, &summary)
+                                .expect("String-to-JSON payload serialization is infallible")
+                        }),
                         details: None,
                     }),
                 },
diff --git a/crates/sdk/src/workflow_context/options.rs b/crates/sdk/src/workflow_context/options.rs
